use std::sync::Mutex;
use std::thread::JoinHandle;
use super::super::reclaim_placement::ReclaimRelease;
use super::{ConnectionSupervisor, SupervisorInner};
use crate::ServerError;
#[derive(Debug, Default)]
pub(super) struct BackgroundStart {
started: bool,
stopped: bool,
worker: Option<JoinHandle<()>>,
failure: Option<String>,
reclaim_releases: Vec<ReclaimRelease>,
}
impl ConnectionSupervisor {
pub(crate) fn retain_reclaim_release(
&self,
release: ReclaimRelease,
) -> Result<(), ServerError> {
let mut start = self
.inner
.background
.lock()
.map_err(|_| refusal("connection background start lock poisoned"))?;
if start.stopped {
return Err(refusal("spine release cannot be registered after shutdown"));
}
start.reclaim_releases.push(release);
drop(start);
Ok(())
}
pub(crate) fn listening(&self) -> Result<(), ServerError> {
let mut start = self
.inner
.background
.lock()
.map_err(|_| refusal("connection background start lock poisoned"))?;
if start.stopped {
return Err(refusal(
"connection background work cannot start after shutdown",
));
}
if !start.started {
start.worker = self.inner.runtime.services_arc().listening();
start.started = true;
}
drop(start);
Ok(())
}
pub(in crate::server) fn finish_store_work(&self) -> Result<(), ServerError> {
self.inner.finish_store_work()
}
pub(in crate::server) fn release_store_holders(&self) -> Result<(), ServerError> {
self.inner.runtime.services().release_store_holders()
}
}
impl SupervisorInner {
fn finish_store_work(&self) -> Result<(), ServerError> {
#[cfg(test)]
let observer = self
.reclaim_stop_observer
.lock()
.map_err(|_| refusal("reclamation stop observer lock poisoned"))?
.take();
#[cfg(test)]
if let Some(observer) = observer {
observer
.send(true)
.map_err(|error| refusal(error.to_string()))?;
}
first_failure([
self.reclaim.finish(&self.scheduler),
join_background(&self.background),
self.runtime.services().finish_background_store_work(),
])
}
}
impl Drop for SupervisorInner {
fn drop(&mut self) {
self.scheduler.shutdown();
if let Err(error) = self.finish_store_work() {
tracing::error!(%error, "connection supervisor store teardown failed");
}
}
}
fn join_background(background: &Mutex<BackgroundStart>) -> Result<(), ServerError> {
let mut start = background
.lock()
.map_err(|_| refusal("connection background join lock poisoned"))?;
start.stopped = true;
if let Some(failure) = &start.failure {
return Err(refusal(failure.clone()));
}
let joined = start.worker.take().map_or(Ok(()), |worker| {
worker
.join()
.map_err(|_| refusal("connection background store work panicked during shutdown"))
});
let mut results = vec![joined];
for release in &mut start.reclaim_releases {
results.push(release.finish());
}
start.reclaim_releases.clear();
let mut first = None;
for result in results {
if let Err(error) = result {
tracing::error!(%error, "background store teardown failed");
if first.is_none() {
first = Some(error);
}
}
}
if let Some(error) = first {
start.failure = Some(error.to_string());
return Err(error);
}
drop(start);
Ok(())
}
fn first_failure<const N: usize>(results: [Result<(), ServerError>; N]) -> Result<(), ServerError> {
let mut first = None;
for result in results {
if let Err(error) = result {
tracing::error!(%error, "connection store work teardown failed");
if first.is_none() {
first = Some(error);
}
}
}
first.map_or(Ok(()), Err)
}
fn refusal(message: impl Into<String>) -> ServerError {
ServerError::ListenerAccept {
message: message.into(),
}
}