Refactor to use dedicated thread for jvm

This commit is contained in:
Bjorn Beishline 2026-01-28 01:46:23 -07:00
parent 21fd37457c
commit 67b9f06381
8 changed files with 428 additions and 184 deletions

View File

@ -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<Mutex<PluginManager>>,
pub command_tx: mpsc::Sender<JvmCommand>,
}
#[with_runtime(global)]
impl EventHandler<PlayerJoinEvent> for PatchBukkitJoinHandler {
fn handle_blocking<'a>(
&self,
server: &Arc<Server>,
_server: &Arc<Server>,
event: &'a mut PlayerJoinEvent,
) -> BoxFuture<'a, ()> {
Box::pin(async {

View File

@ -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<Mutex<PluginManager>>, server: &Arc<Context>) {
pub async fn register_handlers(command_tx: mpsc::Sender<JvmCommand>, server: &Arc<Context>) {
server
.register_event(
Arc::new(PatchBukkitJoinHandler { plugin_manager }),
Arc::new(PatchBukkitJoinHandler { command_tx }),
EventPriority::Highest,
true,
)

View File

@ -1,27 +0,0 @@
use std::path::PathBuf;
use j4rs::{InvocationArg, Jvm, JvmBuilder};
pub fn initialize_jvm(j4rs_folder: &PathBuf) -> Result<Jvm, String> {
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(())
}

View File

@ -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<Result<()>>,
},
JavaCallback {
instance: Instance,
respond_to: oneshot::Sender<Result<()>>,
},
LoadPlugin {
plugin_path: PathBuf,
respond_to: oneshot::Sender<LoadPluginResult>,
},
InstantiateAllPlugins {
respond_to: oneshot::Sender<Result<()>>,
},
EnableAllPlugins {
respond_to: oneshot::Sender<Result<()>>,
},
DisableAllPlugins {
respond_to: oneshot::Sender<Result<()>>,
},
Shutdown {
respond_to: oneshot::Sender<Result<()>>,
},
}

2
rust/src/java/jvm/mod.rs Normal file
View File

@ -0,0 +1,2 @@
pub mod commands;
pub mod worker;

244
rust/src/java/jvm/worker.rs Normal file
View File

@ -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<JvmCommand>,
command_tx: mpsc::Sender<JvmCommand>,
pub plugin_manager: PluginManager,
jvm: Option<j4rs::Jvm>,
}
impl JvmWorker {
pub fn new(
command_tx: mpsc::Sender<JvmCommand>,
command_rx: mpsc::Receiver<JvmCommand>,
) -> 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(())
}

View File

@ -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<Context>) -> 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<Context>,
) -> 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<Context>) -> Result<(), String> {
#[plugin_impl]
pub struct PatchBukkitPlugin {
plugin_manager: Arc<Mutex<PluginManager>>,
pub command_tx: mpsc::Sender<JvmCommand>,
}
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 }
}
}

View File

@ -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",