From 67b9f06381ca3ed9c26490934545004a9013515e Mon Sep 17 00:00:00 2001 From: Bjorn Beishline <75190918+BjornTheProgrammer@users.noreply.github.com> Date: Wed, 28 Jan 2026 01:46:23 -0700 Subject: [PATCH] Refactor to use dedicated thread for jvm --- rust/src/events/join.rs | 8 +- rust/src/events/mod.rs | 8 +- rust/src/java/jvm.rs | 27 ---- rust/src/java/jvm/commands.rs | 41 +++++ rust/src/java/jvm/mod.rs | 2 + rust/src/java/jvm/worker.rs | 244 +++++++++++++++++++++++++++++ rust/src/lib.rs | 280 ++++++++++++++++------------------ rust/src/plugin/manager.rs | 2 +- 8 files changed, 428 insertions(+), 184 deletions(-) delete mode 100644 rust/src/java/jvm.rs create mode 100644 rust/src/java/jvm/commands.rs create mode 100644 rust/src/java/jvm/mod.rs create mode 100644 rust/src/java/jvm/worker.rs diff --git a/rust/src/events/join.rs b/rust/src/events/join.rs index efabab3..82bb330 100644 --- a/rust/src/events/join.rs +++ b/rust/src/events/join.rs @@ -6,19 +6,19 @@ use pumpkin::{ }; use pumpkin_api_macros::with_runtime; use pumpkin_util::text::{TextComponent, color::NamedColor}; -use tokio::sync::Mutex; +use tokio::sync::mpsc; -use crate::plugin::manager::PluginManager; +use crate::java::jvm::commands::JvmCommand; pub struct PatchBukkitJoinHandler { - pub plugin_manager: Arc>, + pub command_tx: mpsc::Sender, } #[with_runtime(global)] impl EventHandler for PatchBukkitJoinHandler { fn handle_blocking<'a>( &self, - server: &Arc, + _server: &Arc, event: &'a mut PlayerJoinEvent, ) -> BoxFuture<'a, ()> { Box::pin(async { diff --git a/rust/src/events/mod.rs b/rust/src/events/mod.rs index aa21f22..d05b16e 100644 --- a/rust/src/events/mod.rs +++ b/rust/src/events/mod.rs @@ -1,16 +1,16 @@ use std::sync::Arc; use pumpkin::plugin::{Context, EventPriority}; -use tokio::sync::Mutex; +use tokio::sync::mpsc; -use crate::{events::join::PatchBukkitJoinHandler, plugin::manager::PluginManager}; +use crate::{events::join::PatchBukkitJoinHandler, java::jvm::commands::JvmCommand}; pub mod join; -pub async fn register_handlers(plugin_manager: Arc>, server: &Arc) { +pub async fn register_handlers(command_tx: mpsc::Sender, server: &Arc) { server .register_event( - Arc::new(PatchBukkitJoinHandler { plugin_manager }), + Arc::new(PatchBukkitJoinHandler { command_tx }), EventPriority::Highest, true, ) diff --git a/rust/src/java/jvm.rs b/rust/src/java/jvm.rs deleted file mode 100644 index 7009b03..0000000 --- a/rust/src/java/jvm.rs +++ /dev/null @@ -1,27 +0,0 @@ -use std::path::PathBuf; - -use j4rs::{InvocationArg, Jvm, JvmBuilder}; - -pub fn initialize_jvm(j4rs_folder: &PathBuf) -> Result { - let jvm = JvmBuilder::new() - .with_base_path(j4rs_folder) - .build() - .map_err(|err| format!("JVM failed to init: {:?}", err))?; - - Ok(jvm) -} - -pub fn setup_patchbukkit_server(jvm: &Jvm) -> Result<(), String> { - let patchbukkit_server = jvm - .create_instance("org.patchbukkit.PatchBukkitServer", InvocationArg::empty()) - .map_err(|err| format!("Failed to create PatchBukkitServer instance: {:?}", err))?; - - jvm.invoke_static( - "org.bukkit.Bukkit", - "setServer", - &[InvocationArg::from(patchbukkit_server)], - ) - .map_err(|err| format!("Failed to set Bukkit server: {:?}", err))?; - - Ok(()) -} diff --git a/rust/src/java/jvm/commands.rs b/rust/src/java/jvm/commands.rs new file mode 100644 index 0000000..295bf08 --- /dev/null +++ b/rust/src/java/jvm/commands.rs @@ -0,0 +1,41 @@ +use std::path::PathBuf; + +use anyhow::Result; +use j4rs::Instance; +use tokio::sync::oneshot; + +pub enum LoadPluginResult { + SuccessfullyLoadedSpigot, + SuccessfullyLoadedPaper, + FailedToLoadSpigotPlugin(anyhow::Error), + FailedToLoadPaperPlugin(anyhow::Error), + FailedToReadConfigurationFile(anyhow::Error), + NoConfigurationFile, +} + +pub enum JvmCommand { + Initialize { + j4rs_path: PathBuf, + respond_to: oneshot::Sender>, + }, + JavaCallback { + instance: Instance, + respond_to: oneshot::Sender>, + }, + LoadPlugin { + plugin_path: PathBuf, + respond_to: oneshot::Sender, + }, + InstantiateAllPlugins { + respond_to: oneshot::Sender>, + }, + EnableAllPlugins { + respond_to: oneshot::Sender>, + }, + DisableAllPlugins { + respond_to: oneshot::Sender>, + }, + Shutdown { + respond_to: oneshot::Sender>, + }, +} diff --git a/rust/src/java/jvm/mod.rs b/rust/src/java/jvm/mod.rs new file mode 100644 index 0000000..1cdb224 --- /dev/null +++ b/rust/src/java/jvm/mod.rs @@ -0,0 +1,2 @@ +pub mod commands; +pub mod worker; diff --git a/rust/src/java/jvm/worker.rs b/rust/src/java/jvm/worker.rs new file mode 100644 index 0000000..0600ba7 --- /dev/null +++ b/rust/src/java/jvm/worker.rs @@ -0,0 +1,244 @@ +use std::path::PathBuf; + +use anyhow::bail; +use j4rs::{Instance, InvocationArg, Jvm, JvmBuilder}; +use tokio::sync::{mpsc, oneshot}; + +use crate::{ + java::{ + jar::read_configs_from_jar, + jvm::commands::{JvmCommand, LoadPluginResult}, + }, + plugin::manager::PluginManager, +}; + +pub struct JvmWorker { + command_rx: mpsc::Receiver, + command_tx: mpsc::Sender, + pub plugin_manager: PluginManager, + jvm: Option, +} + +impl JvmWorker { + pub fn new( + command_tx: mpsc::Sender, + command_rx: mpsc::Receiver, + ) -> Self { + Self { + command_rx, + command_tx, + plugin_manager: PluginManager::new(), + jvm: None, + } + } + + pub fn attach_thread(mut self) { + log::info!("JVM worker thread started"); + + while let Some(command) = self.command_rx.blocking_recv() { + match command { + JvmCommand::Initialize { + j4rs_path, + respond_to, + } => { + let result = self.initialize_jvm(&j4rs_path); + let _ = respond_to.send(result); + } + JvmCommand::JavaCallback { + instance, + respond_to, + } => { + let result = self.process_java_callback(instance); + let _ = respond_to.send(result); + } + JvmCommand::LoadPlugin { + plugin_path, + respond_to, + } => { + let _ = match read_configs_from_jar(&plugin_path) { + Ok(configs) => match configs { + (Some(paper_plugin_config), spigot @ _) => { + match self.plugin_manager.load_paper_plugin( + &plugin_path, + &paper_plugin_config, + &spigot, + ) { + Ok(_) => { + respond_to.send(LoadPluginResult::SuccessfullyLoadedPaper) + } + Err(err) => respond_to + .send(LoadPluginResult::FailedToLoadPaperPlugin(err)), + } + } + (None, Some(spigot)) => { + match self + .plugin_manager + .load_spigot_plugin(&plugin_path, &spigot) + { + Ok(_) => { + respond_to.send(LoadPluginResult::SuccessfullyLoadedSpigot) + } + Err(err) => respond_to + .send(LoadPluginResult::FailedToLoadSpigotPlugin(err)), + } + } + (None, None) => respond_to.send(LoadPluginResult::NoConfigurationFile), + }, + Err(err) => { + respond_to.send(LoadPluginResult::FailedToReadConfigurationFile(err)) + } + }; + } + JvmCommand::InstantiateAllPlugins { respond_to } => { + match self.jvm { + Some(ref jvm) => { + let _ = + respond_to.send(self.plugin_manager.instantiate_all_plugins(jvm)); + } + None => { + unreachable!("Shouldn't be able to instantiate plugins without a JVM") + } + }; + } + JvmCommand::EnableAllPlugins { respond_to } => { + match self.jvm { + Some(ref jvm) => { + let _ = respond_to.send(self.plugin_manager.enable_all_plugins(jvm)); + } + None => { + unreachable!("Shouldn't be able to enable plugins without a JVM") + } + }; + } + JvmCommand::DisableAllPlugins { respond_to } => { + match self.jvm { + Some(ref jvm) => { + let _ = respond_to.send(self.plugin_manager.disable_all_plugins(jvm)); + } + None => { + unreachable!("Shouldn't be able to disable plugins without a JVM") + } + }; + } + JvmCommand::Shutdown { respond_to } => { + let _ = respond_to.send(self.plugin_manager.unload_all_plugins()); + break; + } + } + } + + log::info!("JVM worker thread exited"); + } + + fn process_java_callback(&mut self, instance: Instance) -> anyhow::Result<()> { + let jvm = match self.jvm { + Some(ref jvm) => jvm, + None => bail!("JVM not initialized"), + }; + + let result = jvm.invoke(&instance, "getCallbackName", InvocationArg::empty())?; + + let callback_name: String = jvm.to_rust(result)?; + match callback_name.as_str() { + "REGISTER_EVENT_CALLBACK" => { + let listener_name: String = jvm.to_rust(jvm.invoke( + &instance, + "getArg", + &[&InvocationArg::try_from(0_i32)?], + )?)?; + let listener = jvm.cast( + &jvm.invoke(&instance, "getArg", &[&InvocationArg::try_from(1_i32)?])?, + "org.bukkit.event.Listener", + )?; + let plugin_instance = jvm.cast( + &jvm.invoke(&instance, "getArg", &[&InvocationArg::try_from(2_i32)?])?, + "org.bukkit.plugin.Plugin", + )?; + let plugin_name: String = jvm.to_rust(jvm.invoke( + &plugin_instance, + "getName", + InvocationArg::empty(), + )?)?; + + match self.plugin_manager.plugins.get_mut(&plugin_name) { + Some(rust_plugin) => rust_plugin.listeners.insert(listener_name, listener), + None => todo!(), + }; + } + _ => log::warn!("Received unknown callback: {:?}", callback_name), + } + + Ok(()) + } + + fn initialize_jvm(&mut self, j4rs_path: &PathBuf) -> anyhow::Result<()> { + log::info!("Initializing JVM with path: {:?}", j4rs_path); + + let jvm = JvmBuilder::new().with_base_path(j4rs_path).build()?; + + setup_patchbukkit_server(&jvm)?; + + let callback_instance = + jvm.create_instance("org.patchbukkit.NativeCallbacks", InvocationArg::empty())?; + + let callback_receiver = jvm.init_callback_channel(&callback_instance)?; + + let command_tx = self.command_tx.clone(); + std::thread::Builder::new() + .name("patchbukkit-jvm-callbacks".to_string()) + .spawn(move || { + let thread_runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + + thread_runtime.block_on(async { + loop { + match callback_receiver.rx().recv() { + Ok(instance) => { + let (sender, receiver) = oneshot::channel(); + if command_tx + .blocking_send(JvmCommand::JavaCallback { + instance, + respond_to: sender, + }) + .is_err() + { + log::info!( + "Command channel closed, stopping callback forwarder" + ); + break; + } + + if let Err(e) = receiver.await { + log::error!("Callback failed: {}", e); + } + } + Err(e) => { + log::error!("Java callback channel closed: {}", e); + break; + } + } + } + }); + })?; + + self.jvm = Some(jvm); + + log::info!("JVM initialized successfully"); + Ok(()) + } +} + +pub fn setup_patchbukkit_server(jvm: &Jvm) -> anyhow::Result<()> { + let patchbukkit_server = + jvm.create_instance("org.patchbukkit.PatchBukkitServer", InvocationArg::empty())?; + + jvm.invoke_static( + "org.bukkit.Bukkit", + "setServer", + &[InvocationArg::from(patchbukkit_server)], + )?; + + Ok(()) +} diff --git a/rust/src/lib.rs b/rust/src/lib.rs index b648ac1..59c79f5 100644 --- a/rust/src/lib.rs +++ b/rust/src/lib.rs @@ -1,6 +1,5 @@ use std::sync::Arc; -use j4rs::{InvocationArg, Jvm}; use pumpkin::plugin::Context; use pumpkin_api_macros::{plugin_impl, plugin_method}; @@ -13,162 +12,133 @@ pub mod plugin; use directories::setup_directories; use java::{ jar::discover_jar_files, - jvm::{initialize_jvm, setup_patchbukkit_server}, resources::{cleanup_stale_files, sync_embedded_resources}, }; -use tokio::sync::Mutex; +use tokio::sync::{mpsc, oneshot}; use crate::{ - events::register_handlers, java::jar::read_configs_from_jar, plugin::manager::PluginManager, + events::register_handlers, + java::jvm::{ + commands::{JvmCommand, LoadPluginResult}, + worker::JvmWorker, + }, }; async fn on_load_inner(plugin: &mut PatchBukkitPlugin, server: Arc) -> Result<(), String> { server.init_log(); log::info!("Starting PatchBukkit"); - register_handlers(plugin.plugin_manager.clone(), &server).await; + register_handlers(plugin.command_tx.clone(), &server).await; // Setup directories let dirs = setup_directories(&server)?; // Discover and prepare JAR files let jar_paths = discover_jar_files(&dirs.plugins); - for jar_path in &jar_paths { - match read_configs_from_jar(jar_path) { - Ok(configs) => match configs { - (Some(paper_plugin_config), spigot @ _) => { - match plugin.plugin_manager.lock().await.load_paper_plugin( - jar_path, - &paper_plugin_config, - &spigot, - ) { - Ok(_) => log::info!("Loaded Paper plugin from JAR: {}", jar_path.display()), - Err(err) => log::error!("Failed to load Paper plugin from JAR: {}", err), - } - } - (None, Some(spigot)) => { - match plugin - .plugin_manager - .lock() - .await - .load_spigot_plugin(jar_path, &spigot) - { - Ok(_) => { - log::info!("Loaded Spigot plugin from JAR: {}", jar_path.display()) - } - Err(err) => log::error!("Failed to load Spigot plugin from JAR: {}", err), - } - } - (None, None) => log::warn!( - "Could not find any plugin configuration in JAR: {}", - jar_path.display() - ), - }, - Err(err) => { + for jar_path in jar_paths { + { + let (tx, rx) = oneshot::channel(); + let result = plugin + .command_tx + .send(JvmCommand::LoadPlugin { + plugin_path: jar_path.clone(), + respond_to: tx, + }) + .await; + if let Err(e) = result { log::error!( - "Failed to read configs from PatchBukkit plugin jar: {}", - err + "Failed to send command to load plugin {}: {}", + jar_path.display(), + e ); } - } + match rx.await { + Ok(result) => match result { + LoadPluginResult::SuccessfullyLoadedSpigot => { + log::info!("Loaded Spigot plugin from JAR `{}`", jar_path.display()) + } + LoadPluginResult::SuccessfullyLoadedPaper => { + log::info!("Loaded Paper plugin from JAR `{}`", jar_path.display()) + } + LoadPluginResult::FailedToLoadSpigotPlugin(error) => { + log::error!( + "Failed to load Spigot plugin from JAR `{}` with error: {}", + jar_path.display(), + error + ) + } + LoadPluginResult::FailedToLoadPaperPlugin(error) => { + log::error!( + "Failed to load Paper plugin from JAR `{}` with error: {}", + jar_path.display(), + error + ) + } + LoadPluginResult::FailedToReadConfigurationFile(error) => { + log::error!( + "Failed to read configuration file from JAR `{}`: {}", + jar_path.display(), + error + ) + } + LoadPluginResult::NoConfigurationFile => { + log::warn!( + "No configuration file found for plugin from JAR `{}`", + jar_path.display() + ) + } + }, + Err(e) => log::error!( + "Failed to receive load plugin response for JAR `{}`: {}", + jar_path.display(), + e + ), + } + }; } // Manage embedded resources cleanup_stale_files(&dirs.j4rs); sync_embedded_resources(&dirs.j4rs)?; - // Initialize JVM and PatchBukkit server - let jvm = initialize_jvm(&dirs.j4rs)?; - setup_patchbukkit_server(&jvm)?; + { + let (tx, rx) = oneshot::channel(); + plugin + .command_tx + .send(JvmCommand::Initialize { + j4rs_path: dirs.j4rs, + respond_to: tx, + }) + .await + .map_err(|e| format!("Failed to send command to initialize J4RS: {}", e))?; + rx.await + .map_err(|e| format!("Unable to receive response from J4RS initialization: {}", e))? + .map_err(|e| format!("Failed to initialize all plugins: {}", e))?; + } - let i = jvm - .create_instance("org.patchbukkit.NativeCallbacks", InvocationArg::empty()) - .map_err(|err| err.to_string())?; - let r = jvm.init_callback_channel(&i).unwrap(); + { + let (tx, rx) = oneshot::channel(); + plugin + .command_tx + .send(JvmCommand::InstantiateAllPlugins { respond_to: tx }) + .await + .map_err(|e| format!("Failed to send command to instantiate plugins: {}", e))?; + rx.await + .map_err(|e| format!("Unable to receive response from instantiate plugins: {}", e))? + .map_err(|e| format!("Failed to instantiate all plugins: {}", e))?; + } - let plugin_manager = plugin.plugin_manager.clone(); - tokio::spawn(async move { - loop { - match r.rx().recv() { - Ok(ret) => { - let result: anyhow::Result<()> = (async || { - let jvm = match Jvm::attach_thread() { - Ok(jvm) => jvm, - Err(e) => { - log::error!("Failed to attach callback thread to JVM: {}", e); - return Ok(()); - } - }; - - let result = jvm.invoke(&ret, "getCallbackName", InvocationArg::empty())?; - - let callback_name: String = jvm.to_rust(result)?; - match callback_name.as_str() { - "REGISTER_EVENT_CALLBACK" => { - let listener_name: String = jvm.to_rust(jvm.invoke( - &ret, - "getArg", - &[&InvocationArg::try_from(0_i32)?], - )?)?; - let listener = jvm.cast( - &jvm.invoke( - &ret, - "getArg", - &[&InvocationArg::try_from(1_i32)?], - )?, - "org.bukkit.event.Listener", - )?; - let plugin_instance = jvm.cast( - &jvm.invoke( - &ret, - "getArg", - &[&InvocationArg::try_from(2_i32)?], - )?, - "org.bukkit.plugin.Plugin", - )?; - let plugin_name: String = jvm.to_rust(jvm.invoke( - &plugin_instance, - "getName", - InvocationArg::empty(), - )?)?; - - match plugin_manager.lock().await.plugins.get_mut(&plugin_name) { - Some(rust_plugin) => { - rust_plugin.listeners.insert(listener_name, listener) - } - None => todo!(), - }; - } - _ => log::warn!("Received unknown callback: {:?}", callback_name), - } - - return Ok(()); - })() - .await; - - result.unwrap(); - } - Err(e) => { - log::error!("Callback channel closed: {}", e); - break; - } - } - } - }); - - plugin - .plugin_manager - .lock() - .await - .load_all_plugins(&jvm) - .map_err(|err| format!("Failed to load PatchBukkit plugins: {}", err))?; - - plugin - .plugin_manager - .lock() - .await - .enable_all_plugins(&jvm) - .map_err(|err| format!("Failed to enable PatchBukkit plugins: {}", err))?; + { + let (tx, rx) = oneshot::channel(); + plugin + .command_tx + .send(JvmCommand::EnableAllPlugins { respond_to: tx }) + .await + .map_err(|e| format!("Failed to send command to enable all plugins: {}", e))?; + rx.await + .map_err(|e| format!("Unable to receive response from enable all plugins: {}", e))? + .map_err(|e| format!("Failed to enable all plugins: {}", e))?; + }; Ok(()) } @@ -177,21 +147,29 @@ async fn on_unload_inner( plugin: &mut PatchBukkitPlugin, _server: Arc, ) -> Result<(), String> { - let jvm = Jvm::attach_thread().map_err(|e| format!("Failed to attach thread to JVM: {}", e))?; + { + let (tx, rx) = oneshot::channel(); + plugin + .command_tx + .send(JvmCommand::DisableAllPlugins { respond_to: tx }) + .await + .map_err(|e| format!("Failed to send command to disable all plugins: {}", e))?; + rx.await + .map_err(|e| format!("Unable to receive response from disable all plugins: {}", e))? + .map_err(|e| format!("Failed to disable all plugins: {}", e))?; + } - plugin - .plugin_manager - .lock() - .await - .disable_all_plugins(&jvm) - .map_err(|err| format!("Failed to disable PatchBukkit plugins: {}", err))?; - - plugin - .plugin_manager - .lock() - .await - .unload_all_plugins() - .map_err(|err| format!("Failed to unload PatchBukkit plugins: {}", err))?; + { + let (tx, rx) = oneshot::channel(); + plugin + .command_tx + .send(JvmCommand::Shutdown { respond_to: tx }) + .await + .map_err(|e| format!("Failed to send command to shutdown: {}", e))?; + rx.await + .map_err(|e| format!("Unable to receive response from shutdown: {}", e))? + .map_err(|e| format!("Failed to shutdown: {}", e))?; + } Ok(()) } @@ -208,14 +186,20 @@ async fn on_unload(&mut self, server: Arc) -> Result<(), String> { #[plugin_impl] pub struct PatchBukkitPlugin { - plugin_manager: Arc>, + pub command_tx: mpsc::Sender, } impl PatchBukkitPlugin { pub fn new() -> Self { - PatchBukkitPlugin { - plugin_manager: Arc::new(Mutex::new(PluginManager::new())), - } + let (tx, rx) = mpsc::channel(100); + let command_tx = tx.clone(); + std::thread::Builder::new() + .name("patchbukkit-jvm-worker".to_string()) + .spawn(move || { + JvmWorker::new(command_tx.clone(), rx).attach_thread(); + }) + .unwrap(); + PatchBukkitPlugin { command_tx: tx } } } diff --git a/rust/src/plugin/manager.rs b/rust/src/plugin/manager.rs index 0c0ec83..26b7a5e 100644 --- a/rust/src/plugin/manager.rs +++ b/rust/src/plugin/manager.rs @@ -181,7 +181,7 @@ impl PluginManager { Ok(()) } - pub fn load_all_plugins(&mut self, jvm: &Jvm) -> Result<()> { + pub fn instantiate_all_plugins(&mut self, jvm: &Jvm) -> Result<()> { for (_plugin_name, plugin) in &mut self.plugins { let result = jvm.invoke_static( "org.patchbukkit.loader.PatchBukkitPluginLoader",