diff --git a/.changeset/kind-flowers-relate.md b/.changeset/kind-flowers-relate.md new file mode 100644 index 00000000..cba86abb --- /dev/null +++ b/.changeset/kind-flowers-relate.md @@ -0,0 +1,5 @@ +--- +"python-sdk": minor +--- + +Add shutdown() method for graceful client shutdown. diff --git a/.changeset/loud-lions-confess.md b/.changeset/loud-lions-confess.md new file mode 100644 index 00000000..c00d1647 --- /dev/null +++ b/.changeset/loud-lions-confess.md @@ -0,0 +1,5 @@ +--- +"eppo_core": minor +--- + +Add by-ref graceful shutdown for BackgroundThread. diff --git a/eppo_core/src/background/thread.rs b/eppo_core/src/background/thread.rs index 2f2a6010..c0ad09f8 100644 --- a/eppo_core/src/background/thread.rs +++ b/eppo_core/src/background/thread.rs @@ -1,10 +1,12 @@ +use std::sync::Mutex; + use super::runtime::BackgroundRuntime; /// An owning handle to a background thread running tokio runtime. /// /// When the handle is dropped, the tokio runtime is commanded to exit and the thread shuts down. pub struct BackgroundThread { - join_handle: std::thread::JoinHandle<()>, + join_handle: Mutex>>, runtime: BackgroundRuntime, } @@ -23,11 +25,13 @@ impl BackgroundThread { let join_handle = std::thread::Builder::new() .name("eppo-background".to_owned()) .spawn(move || { + log::info!(target: "eppo", "BackgroundThread: started"); runtime.block_on(wait); + log::info!(target: "eppo", "BackgroundThread: exiting"); })?; Ok(BackgroundThread { - join_handle, + join_handle: Mutex::new(Some(join_handle)), runtime: background_runtime, }) } @@ -38,16 +42,32 @@ impl BackgroundThread { /// Command the associated background thread to exit (without waiting for it to complete). /// - /// Prefer `graceful_shutdown()` if you have the time to wait. + /// Prefer `shutdown()` if you have the time to wait. pub fn kill(&self) { self.runtime.stop(); } /// Command background activities to stop and wait for thread to terminate. - pub fn graceful_shutdown(self) { + pub fn shutdown(&self) { self.runtime.stop(); - let _ = self.join_handle.join(); + let join_handle = { + // scope to keep mutex lock short + let Ok(mut join_handle) = self.join_handle.lock() else { + return; + }; + join_handle.take() + }; + + if let Some(join_handle) = join_handle { + let _ = join_handle.join(); + } + } + + /// Command background activities to stop and wait for thread to terminate. + #[deprecated] + pub fn graceful_shutdown(self) { + self.shutdown(); } } @@ -69,6 +89,6 @@ mod tests { assert_eq!(received, true); - background_thread.graceful_shutdown(); + background_thread.shutdown(); } } diff --git a/eppo_core/src/configuration_fetcher.rs b/eppo_core/src/configuration_fetcher.rs index 3f090042..90ca5715 100644 --- a/eppo_core/src/configuration_fetcher.rs +++ b/eppo_core/src/configuration_fetcher.rs @@ -31,8 +31,10 @@ impl ConfigurationFetcher { pub fn new(config: ConfigurationFetcherConfig) -> ConfigurationFetcher { let builder = reqwest::Client::builder(); let client = match builder.build() { - Err(e) => { panic!("Reqwest client build failed {:?}", e); } - Ok(client) => client + Err(e) => { + panic!("Reqwest client build failed {:?}", e); + } + Ok(client) => client, }; ConfigurationFetcher { diff --git a/eppo_core/src/event_ingestion/event.rs b/eppo_core/src/event_ingestion/event.rs index edfb77b4..12724e9d 100644 --- a/eppo_core/src/event_ingestion/event.rs +++ b/eppo_core/src/event_ingestion/event.rs @@ -15,4 +15,4 @@ pub(super) struct Event { pub event_type: String, pub payload: serde_json::Value, -} \ No newline at end of file +} diff --git a/package-lock.json b/package-lock.json index 63d0cdd0..75647d7b 100644 --- a/package-lock.json +++ b/package-lock.json @@ -22,15 +22,15 @@ } }, "dart-sdk": { - "version": "0.1.0", + "version": "0.1.1", "devDependencies": {} }, "elixir-sdk": { - "version": "0.1.0", + "version": "0.2.3", "devDependencies": {} }, "eppo_core": { - "version": "9.0.0", + "version": "9.2.0", "devDependencies": {} }, "mock-server": { @@ -3362,11 +3362,11 @@ } }, "python-sdk": { - "version": "4.3.0", + "version": "4.3.1", "devDependencies": {} }, "ruby-sdk": { - "version": "3.5.0", + "version": "3.7.2", "devDependencies": {} }, "rust-sdk": { diff --git a/python-sdk/python/eppo_client/_eppo_client.pyi b/python-sdk/python/eppo_client/_eppo_client.pyi index 64062ec1..1b9f974c 100644 --- a/python-sdk/python/eppo_client/_eppo_client.pyi +++ b/python-sdk/python/eppo_client/_eppo_client.pyi @@ -143,6 +143,7 @@ class EppoClient: def set_is_graceful_mode(self, is_graceful_mode: bool): ... def is_initialized(self) -> bool: ... def wait_for_initialization(self) -> None: ... + def shutdown(self) -> None: ... class ContextAttributes: def __new__( @@ -154,7 +155,7 @@ class ContextAttributes: def empty() -> ContextAttributes: ... @staticmethod def from_dict( - attributes: Dict[str, Union[str, int, float, bool, None]] + attributes: Dict[str, Union[str, int, float, bool, None]], ) -> ContextAttributes: ... @property def numeric_attributes(self) -> Dict[str, float]: ... diff --git a/python-sdk/src/client.rs b/python-sdk/src/client.rs index a508a43b..bd24dc12 100644 --- a/python-sdk/src/client.rs +++ b/python-sdk/src/client.rs @@ -469,6 +469,23 @@ impl EppoClient { } } + /// Shutdown the client and wait for the background thread to exit. + /// + /// It is recommended to call this method before exiting the program so the background thread + /// has a chance to cleanup its resources and avoid getting killed, producing scary (but + /// harmless) panic backtrace. + pub fn shutdown(&self, py: Python) { + log::info!(target: "eppo", "shutting down client"); + // Release python GIL before we try blocking and waiting for the background thread to + // join. This is required because pyo3_log in background thread might attempt acquiring GIL, + // which would result in deadlock as GIL is held by the current thread which is waiting for + // background thread to exit. + py.allow_threads(|| { + if let Some(thread) = &self.background_thread { + thread.shutdown(); + } + }) + } /// Returns a set of all flag keys that have been initialized. /// This can be useful to debug the initialization process. /// @@ -699,18 +716,12 @@ impl EppoClient { .call_method1(py, intern!(py, "log_bandit_action"), (event,))?; Ok(()) } - - pub fn shutdown(&self) { - if let Some(thread) = &self.background_thread { - // Using `.kill()` instead of `.shutdown()` here because we don't need to wait for the - // poller thread to exit. - thread.kill(); - } - } } impl Drop for EppoClient { fn drop(&mut self) { - self.shutdown(); + if let Some(thread) = &self.background_thread { + thread.kill(); + } } } diff --git a/python-sdk/src/init.rs b/python-sdk/src/init.rs index 9e674fdc..7ee9af9c 100644 --- a/python-sdk/src/init.rs +++ b/python-sdk/src/init.rs @@ -35,7 +35,7 @@ pub fn init(config: Bound) -> PyResult> { std::mem::replace(&mut *instance, Some(client)) }; if let Some(existing) = existing { - existing.get().shutdown(); + existing.get().shutdown(py); existing.drop_ref(py); } diff --git a/python-sdk/tests/test_client.py b/python-sdk/tests/test_client.py index 0508e218..7636b1e0 100644 --- a/python-sdk/tests/test_client.py +++ b/python-sdk/tests/test_client.py @@ -24,6 +24,15 @@ def test_wait_for_initialization(): assert client.is_initialized() == True +@pytest.mark.rust_only +def test_shutdown(): + client = init("ufc", wait_for_init=False) + + client.shutdown() + # safe to call multiple times: + client.shutdown() + + def test_get_flag_keys_none(): client = init("ufc", wait_for_init=False) assert client.get_flag_keys() == set()