Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -333,7 +333,7 @@ fn limits(options: &KeywordArguments<'_>) -> PyResult<Limits> {
/// Stop every live dispatcher's threads and join them before finalization ends them inside native code.
#[pyfunction]
pub(crate) fn stop_event_exporter_threads(py: Python<'_>) -> PyResult<()> {
signal_interpreter_exit(py)?;
signal_interpreter_exit();
let mut threads: Vec<Bound<'_, PyAny>> = Vec::new();
for thread in imported(py, "threading", "enumerate")?
.call0()?
Expand Down Expand Up @@ -483,7 +483,7 @@ impl Shared {
let threads: Vec<Py<PyAny>> = {
let _held = self.hold(py)?;
loop {
if interpreter_exiting(py)? {
if interpreter_exiting() {
return Ok(());
}
let state = self.locked()?;
Expand Down Expand Up @@ -597,7 +597,7 @@ impl Shared {
Err(error) if error.is_instance(py, &empty) => {}
Err(error) => return Err(error),
}
if interpreter_exiting(py)? {
if interpreter_exiting() {
return Ok(Delivery::Abandoned);
}
}
Expand Down
Original file line number Diff line number Diff line change
@@ -1,30 +1,15 @@
//! The interpreter-exit signal native waits poll so Python daemon threads leave native frames before finalization.

use pyo3::prelude::{Bound, Py, PyAny, PyAnyMethods, PyResult, Python};
use pyo3::sync::PyOnceLock;
use std::sync::atomic::{AtomicBool, Ordering};

use crate::bindings::_helpers::boundary::python_calls::imported;

static EXIT_SIGNAL: PyOnceLock<Py<PyAny>> = PyOnceLock::new();

/// The process-wide `threading.Event` set once interpreter exit begins.
fn exit_signal(py: Python<'_>) -> PyResult<Bound<'_, PyAny>> {
EXIT_SIGNAL
.get_or_try_init(py, || {
imported(py, "threading", "Event")?
.call0()
.map(Bound::unbind)
})
.map(|signal| signal.bind(py).clone())
}
static EXIT_SIGNAL: AtomicBool = AtomicBool::new(false);

/// Whether an exit hook has started stopping native daemon threads.
pub(crate) fn interpreter_exiting(py: Python<'_>) -> PyResult<bool> {
exit_signal(py)?.call_method0("is_set")?.is_truthy()
pub(crate) fn interpreter_exiting() -> bool {
EXIT_SIGNAL.load(Ordering::SeqCst)
}

/// Tell every native wait to stop waiting on Python work that may outlive the interpreter.
pub(crate) fn signal_interpreter_exit(py: Python<'_>) -> PyResult<()> {
exit_signal(py)?.call_method0("set")?;
Ok(())
pub(crate) fn signal_interpreter_exit() {
EXIT_SIGNAL.store(true, Ordering::SeqCst);
}
Original file line number Diff line number Diff line change
Expand Up @@ -122,7 +122,7 @@ fn delivering_thread(py: Python<'_>) -> PyResult<bool> {
/// Stop every capture worker and join it before finalization ends it inside native code.
#[pyfunction]
fn stop_output_capture_threads(py: Python<'_>) -> PyResult<()> {
signal_interpreter_exit(py)?;
signal_interpreter_exit();
let mut threads: Vec<Bound<'_, PyAny>> = Vec::new();
for thread in imported(py, "threading", "enumerate")?
.call0()?
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -215,7 +215,7 @@ pub(crate) fn start_planning(
/// Join SQL test planning threads at exit: finalization ending one inside native planning aborts.
#[pyfunction]
fn join_sql_test_planning_threads(py: Python<'_>) -> PyResult<()> {
signal_interpreter_exit(py)?;
signal_interpreter_exit();
let threads: &PlanningThreads = planning_threads(py)?.get();
py.detach(|| threads.wait(PLANNING_EXIT_JOIN));
Ok(())
Expand Down
7 changes: 6 additions & 1 deletion fensu.toml
Original file line number Diff line number Diff line change
Expand Up @@ -1574,6 +1574,12 @@ rule = "FPRSL105"
path = "crates/sqlbuild-analysis/src/sql_tokens/main/builtin_function_names.rs"
reason = "This test-support entry exists only for the SQL lint tests in sqlbuild-rules, which the analyzer does not count as importers."

[[targets.rust.rule_exceptions]]
rule = "FPRSS130"
path = "crates/sqlbuild-python/src/bindings/_helpers/native_threads/exit_signal.rs"
symbols = ["signal_interpreter_exit"]
reason = "The existing interpreter-exit signal is process-wide and monotonic: every native daemon wait must observe any exit hook before Python finalization. The atomic replaces the same global threading.Event, not per-command lifecycle state."

[targets.rust.rule_options.FPRSR311]
allowed_entries = [
"Cargo.toml",
Expand Down Expand Up @@ -1990,7 +1996,6 @@ pending_budgets = [
"crates/sqlbuild-python/src/bindings/_helpers/model_config/template_data.rs=4",
"crates/sqlbuild-python/src/bindings/_helpers/model_config/typed_config.rs=2",
"crates/sqlbuild-python/src/bindings/_helpers/model_config/unicode_environment.rs=4",
"crates/sqlbuild-python/src/bindings/_helpers/native_threads/exit_signal.rs=4",
"crates/sqlbuild-python/src/bindings/_helpers/output_capture/context.rs=6",
"crates/sqlbuild-python/src/bindings/_helpers/output_capture/dispatcher.rs=33",
"crates/sqlbuild-python/src/bindings/_helpers/output_capture/records.rs=16",
Expand Down
Loading