From 33abab2f9f88410d2a57e95c9ca403485fe268b0 Mon Sep 17 00:00:00 2001 From: segfaultdoctor <17258903+segfaultdoc@users.noreply.github.com> Date: Thu, 19 Oct 2023 15:08:50 -0400 Subject: [PATCH] wip --- runtime-plugin/Cargo.toml | 18 + runtime-plugin/src/lib.rs | 4 + runtime-plugin/src/runtime_plugin.rs | 41 +++ .../src/runtime_plugin_admin_rpc_service.rs | 312 ++++++++++++++++++ runtime-plugin/src/runtime_plugin_manager.rs | 278 ++++++++++++++++ runtime-plugin/src/runtime_plugin_service.rs | 121 +++++++ 6 files changed, 774 insertions(+) create mode 100644 runtime-plugin/Cargo.toml create mode 100644 runtime-plugin/src/lib.rs create mode 100644 runtime-plugin/src/runtime_plugin.rs create mode 100644 runtime-plugin/src/runtime_plugin_admin_rpc_service.rs create mode 100644 runtime-plugin/src/runtime_plugin_manager.rs create mode 100644 runtime-plugin/src/runtime_plugin_service.rs diff --git a/runtime-plugin/Cargo.toml b/runtime-plugin/Cargo.toml new file mode 100644 index 0000000000..fc470d993a --- /dev/null +++ b/runtime-plugin/Cargo.toml @@ -0,0 +1,18 @@ +[package] +name = "solana-runtime-plugin" +version = "1.18.0" +edition = "2021" + +[dependencies] +crossbeam-channel = { workspace = true } +json5 = { workspace = true } +jsonrpc-core = { workspace = true } +jsonrpc-core-client = { workspace = true } +jsonrpc-derive = { workspace = true } +jsonrpc-ipc-server = { workspace = true } +jsonrpc-server-utils = { workspace = true } +libloading = { workspace = true } +log = { workspace = true } +solana-runtime = { workspace = true } +solana-sdk = { workspace = true } +thiserror = { workspace = true } diff --git a/runtime-plugin/src/lib.rs b/runtime-plugin/src/lib.rs new file mode 100644 index 0000000000..477af43c9b --- /dev/null +++ b/runtime-plugin/src/lib.rs @@ -0,0 +1,4 @@ +pub mod runtime_plugin; +pub mod runtime_plugin_admin_rpc_service; +pub mod runtime_plugin_manager; +pub mod runtime_plugin_service; diff --git a/runtime-plugin/src/runtime_plugin.rs b/runtime-plugin/src/runtime_plugin.rs new file mode 100644 index 0000000000..7dc0b95fa4 --- /dev/null +++ b/runtime-plugin/src/runtime_plugin.rs @@ -0,0 +1,41 @@ +use { + solana_runtime::{bank_forks::BankForks, commitment::BlockCommitmentCache}, + std::{ + any::Any, + error, + fmt::Debug, + io, + sync::{atomic::AtomicBool, Arc, RwLock}, + }, + thiserror::Error, +}; + +pub type Result = std::result::Result; + +/// Errors returned by plugin calls +#[derive(Error, Debug)] +pub enum RuntimePluginError { + /// Error opening the configuration file; for example, when the file + /// is not found or when the validator process has no permission to read it. + #[error("Error opening config file. Error detail: ({0}).")] + ConfigFileOpenError(#[from] io::Error), + + /// Any custom error defined by the plugin. + #[error("Plugin-defined custom error. Error message: ({0})")] + Custom(Box), + + #[error("Failed to load a runtime plugin")] + FailedToLoadPlugin(#[from] Box), +} + +pub struct PluginDependencies { + pub bank_forks: Arc>, + pub block_commitment_cache: Arc>, + pub exit: Arc, +} + +pub trait RuntimePlugin: Any + Debug + Send + Sync { + fn name(&self) -> &'static str; + fn on_load(&mut self, config_file: &str, dependencies: PluginDependencies) -> Result<()>; + fn on_unload(&mut self); +} diff --git a/runtime-plugin/src/runtime_plugin_admin_rpc_service.rs b/runtime-plugin/src/runtime_plugin_admin_rpc_service.rs new file mode 100644 index 0000000000..56934b225e --- /dev/null +++ b/runtime-plugin/src/runtime_plugin_admin_rpc_service.rs @@ -0,0 +1,312 @@ +//! RPC interface to dynamically make changes to runtime plugins. + +use { + crossbeam_channel::Sender, + jsonrpc_core::{BoxFuture, ErrorCode, MetaIoHandler, Metadata, Result as JsonRpcResult}, + jsonrpc_derive::rpc, + jsonrpc_ipc_server::{ + tokio::{self, sync::oneshot::channel as oneshot_channel}, + RequestContext, ServerBuilder, + }, + jsonrpc_server_utils::tokio::sync::oneshot::Sender as OneShotSender, + log::*, + solana_sdk::exit::Exit, + std::{ + path::{Path, PathBuf}, + sync::{ + atomic::{AtomicBool, Ordering}, + Arc, RwLock, + }, + }, +}; + +#[derive(Debug)] +pub enum RuntimePluginManagerRpcRequest { + ReloadPlugin { + name: String, + config_file: String, + response_sender: OneShotSender>, + }, + UnloadPlugin { + name: String, + response_sender: OneShotSender>, + }, + LoadPlugin { + config_file: String, + response_sender: OneShotSender>, + }, + ListPlugins { + response_sender: OneShotSender>>, + }, +} + +#[rpc] +pub trait RuntimePluginAdminRpc { + type Metadata; + + #[rpc(meta, name = "reloadPlugin")] + fn reload_plugin( + &self, + meta: Self::Metadata, + name: String, + config_file: String, + ) -> BoxFuture>; + + #[rpc(meta, name = "unloadPlugin")] + fn unload_plugin(&self, meta: Self::Metadata, name: String) -> BoxFuture>; + + #[rpc(meta, name = "loadPlugin")] + fn load_plugin( + &self, + meta: Self::Metadata, + config_file: String, + ) -> BoxFuture>; + + #[rpc(meta, name = "listPlugins")] + fn list_plugins(&self, meta: Self::Metadata) -> BoxFuture>>; +} + +#[derive(Clone)] +pub struct RuntimePluginAdminRpcRequestMetadata { + pub rpc_request_sender: Sender, + pub validator_exit: Arc>, +} + +impl Metadata for RuntimePluginAdminRpcRequestMetadata {} + +/// Start the Runtime Plugin Admin RPC interface. +pub fn run( + ledger_path: &Path, + metadata: RuntimePluginAdminRpcRequestMetadata, + plugin_exit: Arc, +) { + fn rpc_path(ledger_path: &Path) -> PathBuf { + #[cfg(target_family = "windows")] + { + // More information about the wackiness of pipe names over at + // https://docs.microsoft.com/en-us/windows/win32/ipc/pipe-names + if let Some(ledger_filename) = ledger_path.file_name() { + PathBuf::from(format!( + "\\\\.\\pipe\\{}-runtime_plugin_admin.rpc", + ledger_filename.to_string_lossy() + )) + } else { + PathBuf::from("\\\\.\\pipe\\runtime_plugin_admin.rpc") + } + } + #[cfg(not(target_family = "windows"))] + { + ledger_path.join("runtime_plugin_admin.rpc") + } + } + + let rpc_path = rpc_path(ledger_path); + + let event_loop = tokio::runtime::Builder::new_multi_thread() + .thread_name("solRuntimePluginAdminRpc") + .worker_threads(1) + .enable_all() + .build() + .unwrap(); + + std::thread::Builder::new() + .name("solAdminRpc".to_string()) + .spawn(move || { + let mut io = MetaIoHandler::default(); + io.extend_with(RuntimePluginAdminRpcImpl.to_delegate()); + + let validator_exit = metadata.validator_exit.clone(); + + match ServerBuilder::with_meta_extractor(io, move |_req: &RequestContext| { + metadata.clone() + }) + .event_loop_executor(event_loop.handle().clone()) + .start(&format!("{}", rpc_path.display())) + { + Err(e) => { + error!("Unable to start runtime plugin admin rpc service: {e:?}, exiting"); + validator_exit.write().unwrap().exit(); + } + Ok(server) => { + info!("started runtime plugin admin rpc service!"); + let close_handle = server.close_handle(); + let c_plugin_exit = plugin_exit.clone(); + validator_exit + .write() + .unwrap() + .register_exit(Box::new(move || { + close_handle.close(); + c_plugin_exit.store(true, Ordering::Relaxed); + })); + + server.wait(); + plugin_exit.store(true, Ordering::Relaxed); + } + } + }) + .unwrap(); +} + +pub struct RuntimePluginAdminRpcImpl; +impl RuntimePluginAdminRpc for RuntimePluginAdminRpcImpl { + type Metadata = RuntimePluginAdminRpcRequestMetadata; + + fn reload_plugin( + &self, + meta: Self::Metadata, + name: String, + config_file: String, + ) -> BoxFuture> { + Box::pin(async move { + let (response_sender, response_receiver) = oneshot_channel(); + + if meta + .rpc_request_sender + .send(RuntimePluginManagerRpcRequest::ReloadPlugin { + name, + config_file, + response_sender, + }) + .is_err() + { + error!("rpc_request_sender channel closed, exiting"); + meta.validator_exit.write().unwrap().exit(); + + return Err(jsonrpc_core::Error { + code: ErrorCode::InternalError, + message: "Internal channel disconnected while sending the request".to_string(), + data: None, + }); + } + + match response_receiver.await { + Err(_) => { + error!("response_receiver channel closed, exiting"); + meta.validator_exit.write().unwrap().exit(); + Err(jsonrpc_core::Error { + code: ErrorCode::InternalError, + message: "Internal channel disconnected while awaiting the response" + .to_string(), + data: None, + }) + } + Ok(resp) => resp, + } + }) + } + + fn unload_plugin(&self, meta: Self::Metadata, name: String) -> BoxFuture> { + Box::pin(async move { + let (response_sender, response_receiver) = oneshot_channel(); + + if meta + .rpc_request_sender + .send(RuntimePluginManagerRpcRequest::UnloadPlugin { + name, + response_sender, + }) + .is_err() + { + error!("rpc_request_sender channel closed, exiting"); + meta.validator_exit.write().unwrap().exit(); + + return Err(jsonrpc_core::Error { + code: ErrorCode::InternalError, + message: "Internal channel disconnected while sending the request".to_string(), + data: None, + }); + } + + match response_receiver.await { + Err(_) => { + error!("response_receiver channel closed, exiting"); + meta.validator_exit.write().unwrap().exit(); + Err(jsonrpc_core::Error { + code: ErrorCode::InternalError, + message: "Internal channel disconnected while awaiting the response" + .to_string(), + data: None, + }) + } + Ok(resp) => resp, + } + }) + } + + fn load_plugin( + &self, + meta: Self::Metadata, + config_file: String, + ) -> BoxFuture> { + Box::pin(async move { + let (response_sender, response_receiver) = oneshot_channel(); + + if meta + .rpc_request_sender + .send(RuntimePluginManagerRpcRequest::LoadPlugin { + config_file, + response_sender, + }) + .is_err() + { + error!("rpc_request_sender channel closed, exiting"); + meta.validator_exit.write().unwrap().exit(); + + return Err(jsonrpc_core::Error { + code: ErrorCode::InternalError, + message: "Internal channel disconnected while sending the request".to_string(), + data: None, + }); + } + + match response_receiver.await { + Err(_) => { + error!("response_receiver channel closed, exiting"); + meta.validator_exit.write().unwrap().exit(); + Err(jsonrpc_core::Error { + code: ErrorCode::InternalError, + message: "Internal channel disconnected while awaiting the response" + .to_string(), + data: None, + }) + } + Ok(resp) => resp, + } + }) + } + + fn list_plugins(&self, meta: Self::Metadata) -> BoxFuture>> { + Box::pin(async move { + let (response_sender, response_receiver) = oneshot_channel(); + + if meta + .rpc_request_sender + .send(RuntimePluginManagerRpcRequest::ListPlugins { response_sender }) + .is_err() + { + error!("rpc_request_sender channel closed, exiting"); + meta.validator_exit.write().unwrap().exit(); + + return Err(jsonrpc_core::Error { + code: ErrorCode::InternalError, + message: "Internal channel disconnected while sending the request".to_string(), + data: None, + }); + } + + match response_receiver.await { + Err(_) => { + error!("response_receiver channel closed, exiting"); + meta.validator_exit.write().unwrap().exit(); + Err(jsonrpc_core::Error { + code: ErrorCode::InternalError, + message: "Internal channel disconnected while awaiting the response" + .to_string(), + data: None, + }) + } + Ok(resp) => resp, + } + }) + } +} diff --git a/runtime-plugin/src/runtime_plugin_manager.rs b/runtime-plugin/src/runtime_plugin_manager.rs new file mode 100644 index 0000000000..0f1653fa4c --- /dev/null +++ b/runtime-plugin/src/runtime_plugin_manager.rs @@ -0,0 +1,278 @@ +use { + crate::runtime_plugin::{PluginDependencies, RuntimePlugin}, + jsonrpc_core::{serde_json, ErrorCode, Result as JsonRpcResult}, + libloading::Library, + log::*, + solana_runtime::{bank_forks::BankForks, commitment::BlockCommitmentCache}, + std::{ + fs::File, + io::Read, + path::{Path, PathBuf}, + sync::{atomic::AtomicBool, Arc, RwLock}, + }, +}; + +#[derive(thiserror::Error, Debug)] +pub enum RuntimePluginManagerError { + #[error("Cannot open the the plugin config file")] + CannotOpenConfigFile(String), + + #[error("Cannot read the the plugin config file")] + CannotReadConfigFile(String), + + #[error("The config file is not in a valid Json format")] + InvalidConfigFileFormat(String), + + #[error("Plugin library path is not specified in the config file")] + LibPathNotSet, + + #[error("Invalid plugin path")] + InvalidPluginPath, + + #[error("Cannot load plugin shared library")] + PluginLoadError(String), + + #[error("The runtime plugin {0} is already loaded shared library")] + PluginAlreadyLoaded(String), + + #[error("The RuntimePlugin on_load method failed")] + PluginStartError(String), +} + +pub struct RuntimePluginManager { + plugins: Vec>, + libs: Vec, + bank_forks: Arc>, + block_commitment_cache: Arc>, + exit: Arc, +} + +impl RuntimePluginManager { + pub fn new( + bank_forks: Arc>, + block_commitment_cache: Arc>, + exit: Arc, + ) -> Self { + Self { + plugins: vec![], + libs: vec![], + bank_forks, + block_commitment_cache, + exit, + } + } + + /// This method allows dynamic loading of a runtime plugin. + /// Adds to the existing list of loaded plugins. + pub(crate) fn load_plugin( + &mut self, + plugin_config_path: impl AsRef, + ) -> JsonRpcResult { + // First load plugin + let (mut new_plugin, new_lib, config_file) = + load_plugin_from_config(plugin_config_path.as_ref()).map_err(|e| { + jsonrpc_core::Error { + code: ErrorCode::InvalidRequest, + message: format!("Failed to load plugin: {e}"), + data: None, + } + })?; + + // Then see if a plugin with this name already exists, if so return Err. + let name = new_plugin.name(); + if self.plugins.iter().any(|plugin| name.eq(plugin.name())) { + return Err(jsonrpc_core::Error { + code: ErrorCode::InvalidRequest, + message: format!( + "There already exists a plugin named {} loaded. Did not load requested plugin", + name, + ), + data: None, + }); + } + + new_plugin + .on_load( + config_file, + PluginDependencies { + bank_forks: self.bank_forks.clone(), + block_commitment_cache: self.block_commitment_cache.clone(), + exit: self.exit.clone(), + }, + ) + .map_err(|on_load_err| jsonrpc_core::Error { + code: ErrorCode::InvalidRequest, + message: format!( + "on_load method of plugin {} failed: {on_load_err}", + new_plugin.name() + ), + data: None, + })?; + + self.plugins.push(new_plugin); + self.libs.push(new_lib); + + Ok(name.to_string()) + } + + /// Unloads the plugins and loaded plugin libraries, making sure to fire + /// their `on_plugin_unload()` methods so they can do any necessary cleanup. + pub(crate) fn unload_all_plugins(&mut self) { + (0..self.plugins.len()).for_each(|idx| { + self.try_drop_plugin(idx); + }); + } + + pub(crate) fn unload_plugin(&mut self, name: &str) -> JsonRpcResult<()> { + // Check if any plugin names match this one + let idx = if let Some(idx) = self + .plugins + .iter() + .position(|plugin| plugin.name().eq(name)) + { + idx + } else { + // If we don't find one return an error + return Err(jsonrpc_core::error::Error { + code: ErrorCode::InvalidRequest, + message: String::from("The plugin you requested to unload is not loaded"), + data: None, + }); + }; + + // Unload and drop plugin and lib + self.try_drop_plugin(idx); + + Ok(()) + } + + /// Reloads an existing plugin. + pub(crate) fn reload_plugin(&mut self, name: &str, config_file: &str) -> JsonRpcResult<()> { + // Check if any plugin names match this one + let idx = if let Some(idx) = self + .plugins + .iter() + .position(|plugin| plugin.name().eq(name)) + { + idx + } else { + // If we don't find one return an error + return Err(jsonrpc_core::error::Error { + code: ErrorCode::InvalidRequest, + message: String::from("The plugin you requested to reload is not loaded"), + data: None, + }); + }; + + self.try_drop_plugin(idx); + + // Try to load plugin, library + // SAFETY: It is up to the validator to ensure this is a valid plugin library. + let (mut new_plugin, new_lib, new_parsed_config_file) = + load_plugin_from_config(config_file.as_ref()).map_err(|err| jsonrpc_core::Error { + code: ErrorCode::InvalidRequest, + message: err.to_string(), + data: None, + })?; + + // Attempt to on_load with new plugin + return match new_plugin.on_load( + new_parsed_config_file, + PluginDependencies { + bank_forks: self.bank_forks.clone(), + block_commitment_cache: self.block_commitment_cache.clone(), + exit: self.exit.clone(), + }, + ) { + // On success, push plugin and library + Ok(()) => { + self.plugins.push(new_plugin); + self.libs.push(new_lib); + Ok(()) + } + // On failure, return error + Err(err) => Err(jsonrpc_core::error::Error { + code: ErrorCode::InvalidRequest, + message: format!( + "Failed to start new plugin (previous plugin was dropped!): {err}" + ), + data: None, + }), + }; + } + + pub(crate) fn list_plugins(&self) -> JsonRpcResult> { + Ok(self.plugins.iter().map(|p| p.name().to_owned()).collect()) + } + + fn try_drop_plugin(&mut self, idx: usize) { + if idx < self.plugins.len() { + let mut plugin = self.plugins.remove(idx); + let _current_lib = self.libs.remove(idx); + plugin.on_unload(); + } else { + error!("failed to drop plugin: index {idx} out of bounds"); + } + } +} + +fn load_plugin_from_config( + plugin_config_path: &Path, +) -> Result<(Box, Library, &str), RuntimePluginManagerError> { + type PluginConstructor = unsafe fn() -> *mut dyn RuntimePlugin; + use libloading::Symbol; + + let mut file = match File::open(plugin_config_path) { + Ok(file) => file, + Err(err) => { + return Err(RuntimePluginManagerError::CannotOpenConfigFile(format!( + "Failed to open the plugin config file {plugin_config_path:?}, error: {err:?}" + ))); + } + }; + + let mut contents = String::new(); + if let Err(err) = file.read_to_string(&mut contents) { + return Err(RuntimePluginManagerError::CannotReadConfigFile(format!( + "Failed to read the plugin config file {plugin_config_path:?}, error: {err:?}" + ))); + } + + let result: serde_json::Value = match json5::from_str(&contents) { + Ok(value) => value, + Err(err) => { + return Err(RuntimePluginManagerError::InvalidConfigFileFormat(format!( + "The config file {plugin_config_path:?} is not in a valid Json5 format, error: {err:?}" + ))); + } + }; + + let libpath = result["libpath"] + .as_str() + .ok_or(RuntimePluginManagerError::LibPathNotSet)?; + let mut libpath = PathBuf::from(libpath); + if libpath.is_relative() { + let config_dir = plugin_config_path.parent().ok_or_else(|| { + RuntimePluginManagerError::CannotOpenConfigFile(format!( + "Failed to resolve parent of {plugin_config_path:?}", + )) + })?; + libpath = config_dir.join(libpath); + } + + let config_file = plugin_config_path + .as_os_str() + .to_str() + .ok_or(RuntimePluginManagerError::InvalidPluginPath)?; + + let (plugin, lib) = unsafe { + let lib = Library::new(libpath) + .map_err(|e| RuntimePluginManagerError::PluginLoadError(e.to_string()))?; + let constructor: Symbol = lib + .get(b"_create_plugin") + .map_err(|e| RuntimePluginManagerError::PluginLoadError(e.to_string()))?; + (Box::from_raw(constructor()), lib) + }; + + Ok((plugin, lib, config_file)) +} diff --git a/runtime-plugin/src/runtime_plugin_service.rs b/runtime-plugin/src/runtime_plugin_service.rs new file mode 100644 index 0000000000..0e80650656 --- /dev/null +++ b/runtime-plugin/src/runtime_plugin_service.rs @@ -0,0 +1,121 @@ +use { + crate::{ + runtime_plugin::RuntimePluginError, + runtime_plugin_admin_rpc_service::RuntimePluginManagerRpcRequest, + runtime_plugin_manager::RuntimePluginManager, + }, + crossbeam_channel::Receiver, + log::{error, info}, + solana_runtime::{bank_forks::BankForks, commitment::BlockCommitmentCache}, + std::{ + path::PathBuf, + sync::{ + atomic::{AtomicBool, Ordering}, + Arc, RwLock, + }, + thread::{self, JoinHandle}, + time::Duration, + }, +}; + +pub struct RuntimePluginService { + plugin_manager: Arc>, + rpc_thread: JoinHandle<()>, +} + +impl RuntimePluginService { + pub fn start( + plugin_config_files: &[PathBuf], + rpc_receiver: Receiver, + bank_forks: Arc>, + block_commitment_cache: Arc>, + exit: Arc, + ) -> Result { + let mut plugin_manager = + RuntimePluginManager::new(bank_forks, block_commitment_cache, exit.clone()); + + for config in plugin_config_files { + let name = plugin_manager + .load_plugin(config) + .map_err(|e| RuntimePluginError::FailedToLoadPlugin(e.into()))?; + info!("Loaded Runtime Plugin: {name}"); + } + + let plugin_manager = Arc::new(RwLock::new(plugin_manager)); + let rpc_thread = + Self::start_rpc_request_handler(rpc_receiver, plugin_manager.clone(), exit.clone()); + + Ok(Self { + plugin_manager, + rpc_thread, + }) + } + + pub fn join(self) { + self.rpc_thread.join().unwrap(); + self.plugin_manager.write().unwrap().unload_all_plugins(); + } + + fn start_rpc_request_handler( + rpc_receiver: Receiver, + plugin_manager: Arc>, + exit: Arc, + ) -> JoinHandle<()> { + thread::Builder::new() + .name("solRuntimePluginRpc".to_string()) + .spawn(move || { + const TIMEOUT: Duration = Duration::from_secs(3); + while !exit.load(Ordering::Relaxed) { + if let Ok(request) = rpc_receiver.recv_timeout(TIMEOUT) { + match request { + RuntimePluginManagerRpcRequest::ListPlugins { response_sender } => { + let plugin_list = plugin_manager.read().unwrap().list_plugins(); + if response_sender.send(plugin_list).is_err() { + error!("response_sender channel disconnected"); + return; + } + } + RuntimePluginManagerRpcRequest::ReloadPlugin { + ref name, + ref config_file, + response_sender, + } => { + let reload_result = plugin_manager + .write() + .unwrap() + .reload_plugin(name, config_file); + if response_sender.send(reload_result).is_err() { + error!("response_sender channel disconnected"); + return; + } + } + RuntimePluginManagerRpcRequest::LoadPlugin { + ref config_file, + response_sender, + } => { + let load_result = + plugin_manager.write().unwrap().load_plugin(config_file); + if response_sender.send(load_result).is_err() { + error!("response_sender channel disconnected"); + return; + } + } + RuntimePluginManagerRpcRequest::UnloadPlugin { + ref name, + response_sender, + } => { + let unload_result = + plugin_manager.write().unwrap().unload_plugin(name); + if response_sender.send(unload_result).is_err() { + error!("response_sender channel disconnected"); + return; + } + } + } + } + } + plugin_manager.write().unwrap().unload_all_plugins(); + }) + .unwrap() + } +}