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<T> = std::result::Result<T, RuntimePluginError>;
+
+/// 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<dyn error::Error + Send + Sync>),
+
+    #[error("Failed to load a runtime plugin")]
+    FailedToLoadPlugin(#[from] Box<dyn std::error::Error>),
+}
+
+pub struct PluginDependencies {
+    pub bank_forks: Arc<RwLock<BankForks>>,
+    pub block_commitment_cache: Arc<RwLock<BlockCommitmentCache>>,
+    pub exit: Arc<AtomicBool>,
+}
+
+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<JsonRpcResult<()>>,
+    },
+    UnloadPlugin {
+        name: String,
+        response_sender: OneShotSender<JsonRpcResult<()>>,
+    },
+    LoadPlugin {
+        config_file: String,
+        response_sender: OneShotSender<JsonRpcResult<String>>,
+    },
+    ListPlugins {
+        response_sender: OneShotSender<JsonRpcResult<Vec<String>>>,
+    },
+}
+
+#[rpc]
+pub trait RuntimePluginAdminRpc {
+    type Metadata;
+
+    #[rpc(meta, name = "reloadPlugin")]
+    fn reload_plugin(
+        &self,
+        meta: Self::Metadata,
+        name: String,
+        config_file: String,
+    ) -> BoxFuture<JsonRpcResult<()>>;
+
+    #[rpc(meta, name = "unloadPlugin")]
+    fn unload_plugin(&self, meta: Self::Metadata, name: String) -> BoxFuture<JsonRpcResult<()>>;
+
+    #[rpc(meta, name = "loadPlugin")]
+    fn load_plugin(
+        &self,
+        meta: Self::Metadata,
+        config_file: String,
+    ) -> BoxFuture<JsonRpcResult<String>>;
+
+    #[rpc(meta, name = "listPlugins")]
+    fn list_plugins(&self, meta: Self::Metadata) -> BoxFuture<JsonRpcResult<Vec<String>>>;
+}
+
+#[derive(Clone)]
+pub struct RuntimePluginAdminRpcRequestMetadata {
+    pub rpc_request_sender: Sender<RuntimePluginManagerRpcRequest>,
+    pub validator_exit: Arc<RwLock<Exit>>,
+}
+
+impl Metadata for RuntimePluginAdminRpcRequestMetadata {}
+
+/// Start the Runtime Plugin Admin RPC interface.
+pub fn run(
+    ledger_path: &Path,
+    metadata: RuntimePluginAdminRpcRequestMetadata,
+    plugin_exit: Arc<AtomicBool>,
+) {
+    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<JsonRpcResult<()>> {
+        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<JsonRpcResult<()>> {
+        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<JsonRpcResult<String>> {
+        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<JsonRpcResult<Vec<String>>> {
+        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<Box<dyn RuntimePlugin>>,
+    libs: Vec<Library>,
+    bank_forks: Arc<RwLock<BankForks>>,
+    block_commitment_cache: Arc<RwLock<BlockCommitmentCache>>,
+    exit: Arc<AtomicBool>,
+}
+
+impl RuntimePluginManager {
+    pub fn new(
+        bank_forks: Arc<RwLock<BankForks>>,
+        block_commitment_cache: Arc<RwLock<BlockCommitmentCache>>,
+        exit: Arc<AtomicBool>,
+    ) -> 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<Path>,
+    ) -> JsonRpcResult<String /* plugin name */> {
+        // 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<Vec<String>> {
+        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<dyn RuntimePlugin>, 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<PluginConstructor> = 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<RwLock<RuntimePluginManager>>,
+    rpc_thread: JoinHandle<()>,
+}
+
+impl RuntimePluginService {
+    pub fn start(
+        plugin_config_files: &[PathBuf],
+        rpc_receiver: Receiver<RuntimePluginManagerRpcRequest>,
+        bank_forks: Arc<RwLock<BankForks>>,
+        block_commitment_cache: Arc<RwLock<BlockCommitmentCache>>,
+        exit: Arc<AtomicBool>,
+    ) -> Result<Self, RuntimePluginError> {
+        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<RuntimePluginManagerRpcRequest>,
+        plugin_manager: Arc<RwLock<RuntimePluginManager>>,
+        exit: Arc<AtomicBool>,
+    ) -> 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()
+    }
+}