Merge pull request #68 from hzxuzhonghu/cleanup-launch

runtime renaming

Author: kmesh-bot

.gitignore

@@ -4,3 +4,4 @@
 *~
 *.html
 *.log.*
+.vscode

Cargo.lock

@@ -71,6 +71,7 @@ typed-builder = "0.18.2"
 url.workspace = true
 uuid = { version = "1.17.0", features = ["v4"] }
 x509-parser = { version = "0.17", features = ["default"] }
+tokio-util = "0.7.16"
 
 
 [dev-dependencies]

orion-lib/Cargo.toml

@@ -131,13 +131,16 @@ pub fn new_configuration_channel(capacity: usize) -> (ConfigurationSenders, Conf
 
 /// Start the listeners manager directly without spawning a background task.
 /// Caller must be inside a Tokio runtime and await this async function.
-pub async fn start_listener_manager(configuration_receivers: ConfigurationReceivers) -> Result<()> {
+pub async fn start_listener_manager(
+    configuration_receivers: ConfigurationReceivers,
+    ct: tokio_util::sync::CancellationToken,
+) -> Result<()> {
     let ConfigurationReceivers { listener_configuration_receiver, route_configuration_receiver } =
         configuration_receivers;
 
     tracing::debug!("listeners manager starting");
     let mgr = ListenersManager::new(listener_configuration_receiver, route_configuration_receiver);
-    mgr.start().await.map_err(|err| {
+    mgr.start(ct).await.map_err(|err| {
         tracing::warn!(error = %err, "listeners manager exited with error");
         err
     })?;

orion-lib/src/lib.rs

@@ -57,26 +57,30 @@ impl ListenerInfo {
 }
 
 pub struct ListenersManager {
-    configuration_channel: mpsc::Receiver<ListenerConfigurationChange>,
+    listener_configuration_channel: mpsc::Receiver<ListenerConfigurationChange>,
     route_configuration_channel: mpsc::Receiver<RouteConfigurationChange>,
     listener_handles: BTreeMap<&'static str, ListenerInfo>,
 }
 
 impl ListenersManager {
     pub fn new(
-        configuration_channel: mpsc::Receiver<ListenerConfigurationChange>,
+        listener_configuration_channel: mpsc::Receiver<ListenerConfigurationChange>,
         route_configuration_channel: mpsc::Receiver<RouteConfigurationChange>,
     ) -> Self {
-        ListenersManager { configuration_channel, route_configuration_channel, listener_handles: BTreeMap::new() }
+        ListenersManager {
+            listener_configuration_channel,
+            route_configuration_channel,
+            listener_handles: BTreeMap::new(),
+        }
     }
 
-    pub async fn start(mut self) -> Result<()> {
+    pub async fn start(mut self, ct: tokio_util::sync::CancellationToken) -> Result<()> {
         let (tx_secret_updates, _) = broadcast::channel(16);
         let (tx_route_updates, _) = broadcast::channel(16);
-
+        // TODO: create child token for each listener?
         loop {
             tokio::select! {
-                Some(listener_configuration_change) = self.configuration_channel.recv() => {
+                Some(listener_configuration_change) = self.listener_configuration_channel.recv() => {
                     match listener_configuration_change {
                         ListenerConfigurationChange::Added(boxed) => {
                             let (factory, listener_conf) = *boxed;
@@ -112,9 +116,9 @@ impl ListenersManager {
                         warn!("Internal problem when updating a route: {e}");
                     }
                 },
-                else => {
-                    warn!("All listener manager channels are closed...exiting");
-                    return Err("All listener manager channels are closed...exiting".into());
+                _ = ct.cancelled() => {
+                    warn!("Listener manager exiting");
+                    return Ok(());
                 }
             }
         }

orion-lib/src/listeners/listeners_manager.rs

@@ -34,6 +34,8 @@ parking_lot                    = "0.12.3"
 tokio.workspace                = true
 tower.workspace                = true
 tracing.workspace              = true
+pingora-timeout                = "0.3.0"
+
 
 axum = "0.8.1"
 compact_str.workspace = true
@@ -52,6 +54,7 @@ tracing-subscriber = { workspace = true, features = [
   "registry",
   "std",
 ] }
+tokio-util = "0.7.16"
 
 [target.'cfg(not(target_env = "msvc"))'.dependencies]
 tikv-jemallocator = { version = "0.6", optional = true }
@@ -65,5 +68,8 @@ axum-test              = "17.2.0"
 orion-data-plane-api.workspace = true
 tracing-test.workspace = true
 
+[target.'cfg(unix)'.dev-dependencies]
+libc = "0.2"
+
 [lints]
 workspace = true

orion-proxy/Cargo.toml

@@ -23,6 +23,7 @@ mod admin;
 mod core_affinity;
 mod proxy;
 mod runtime;
+mod signal;
 mod xds_configurator;
 
 pub fn run() -> Result<()> {

orion-proxy/src/lib.rs

@@ -26,7 +26,8 @@ static GLOBAL: Jemalloc = Jemalloc;
 #[global_allocator]
 static ALLOC: dhat::Alloc = dhat::Alloc;
 
-fn main() -> orion_error::Result<()> {
+#[tokio::main]
+async fn main() -> orion_error::Result<()> {
     #[cfg(all(feature = "dhat-heap", not(feature = "jemalloc")))]
     let _profiler = dhat::Profiler::new_heap();
     orion_proxy::run()