use std::collections::VecDeque;
use std::future::Future;
use std::panic::{catch_unwind, AssertUnwindSafe};
use std::path::Path;
use std::pin::Pin;
use std::sync::{Arc, Condvar, Mutex, MutexGuard, OnceLock, PoisonError};
use std::task::{Context, Poll, Waker};
use std::thread::JoinHandle;
use std::time::Instant;
use bacnet_types::enums::{ErrorClass, ErrorCode, PropertyIdentifier};
use bacnet_types::error::Error;
use bacnet_types::primitives::PropertyValue;
pub(crate) mod file;
pub(crate) mod staged;
#[cfg(not(test))]
pub(crate) const STAGED_WRITE_LIFETIME: std::time::Duration = std::time::Duration::from_secs(10);
#[cfg(test)]
pub(crate) const STAGED_WRITE_LIFETIME: std::time::Duration = std::time::Duration::from_millis(300);
fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
mutex.lock().unwrap_or_else(PoisonError::into_inner)
}
#[derive(Default)]
pub(crate) struct Event {
state: Mutex<EventState>,
happened: Condvar,
}
#[derive(Default)]
struct EventState {
done: bool,
wakers: Vec<Waker>,
}
impl Event {
pub(crate) fn set(&self) {
let wakers = {
let mut state = lock(&self.state);
state.done = true;
std::mem::take(&mut state.wakers)
};
self.happened.notify_all();
for waker in wakers {
waker.wake();
}
}
pub(crate) fn is_set(&self) -> bool {
lock(&self.state).done
}
fn wait_blocking(&self) {
let mut state = lock(&self.state);
while !state.done {
state = self
.happened
.wait(state)
.unwrap_or_else(PoisonError::into_inner);
}
}
}
#[must_use = "a SaveWait does nothing unless awaited"]
#[derive(Clone)]
pub struct SaveWait {
event: Arc<Event>,
}
impl SaveWait {
pub(crate) fn new(event: Arc<Event>) -> Self {
Self { event }
}
pub fn is_ready(&self) -> bool {
self.event.is_set()
}
pub fn block(&self) {
self.event.wait_blocking();
}
}
impl std::fmt::Debug for SaveWait {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("SaveWait")
.field("ready", &self.is_ready())
.finish()
}
}
impl Future for SaveWait {
type Output = ();
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
let mut state = lock(&self.event.state);
if state.done {
return Poll::Ready(());
}
if !state.wakers.iter().any(|waker| waker.will_wake(cx.waker())) {
state.wakers.push(cx.waker().clone());
}
Poll::Pending
}
}
#[derive(Clone, Default)]
pub(crate) struct SaveTicket {
slot: Arc<TicketSlot>,
}
#[derive(Default)]
struct TicketSlot {
outcome: Mutex<Outcome>,
finished_at: OnceLock<Instant>,
done: Arc<Event>,
}
#[derive(Default)]
enum Outcome {
#[default]
Pending,
Done(Result<(), Error>),
Taken(bool),
}
impl SaveTicket {
fn complete(&self, result: Result<(), Error>) {
*lock(&self.slot.outcome) = Outcome::Done(result);
let _ = self.slot.finished_at.set(Instant::now());
self.slot.done.set();
}
pub(crate) fn is_done(&self) -> bool {
self.slot.done.is_set()
}
pub(crate) fn wait(&self) -> SaveWait {
SaveWait::new(Arc::clone(&self.slot.done))
}
pub(crate) fn issued(&self, wait: &SaveWait) -> bool {
Arc::ptr_eq(&self.slot.done, &wait.event)
}
pub(crate) fn outlived_at(&self, now: Instant) -> bool {
self.slot.finished_at.get().is_some_and(|finished| {
now.saturating_duration_since(*finished) >= STAGED_WRITE_LIFETIME
})
}
pub(crate) fn outlived(&self) -> bool {
self.outlived_at(Instant::now())
}
pub(crate) fn succeeded(&self) -> Option<bool> {
match &*lock(&self.slot.outcome) {
Outcome::Pending => None,
Outcome::Done(result) => Some(result.is_ok()),
Outcome::Taken(ok) => Some(*ok),
}
}
pub(crate) fn take_outcome(&self) -> Result<(), Error> {
self.slot.done.wait_blocking();
let mut outcome = lock(&self.slot.outcome);
match std::mem::take(&mut *outcome) {
Outcome::Done(result) => {
*outcome = Outcome::Taken(result.is_ok());
result
}
Outcome::Taken(ok) => {
*outcome = Outcome::Taken(ok);
if ok {
Ok(())
} else {
Err(Error::Encoding(
"save failed; its error was reported".into(),
))
}
}
Outcome::Pending => unreachable!("the event is set only after the outcome"),
}
}
}
type SaveFn<S> = dyn Fn(&S) -> Result<(), Error> + Send + Sync;
type DoneFn<S> = dyn Fn(&S, &Result<(), Error>) + Send + Sync;
pub(crate) struct SaveWriter<S: Send + 'static> {
shared: Arc<Shared<S>>,
thread: Option<AssertUnwindSafe<JoinHandle<()>>>,
name: String,
}
struct Shared<S> {
queue: Mutex<Queue<S>>,
work: Condvar,
save: AssertUnwindSafe<Box<SaveFn<S>>>,
done: AssertUnwindSafe<Box<DoneFn<S>>>,
}
struct Queue<S> {
jobs: VecDeque<Job<S>>,
running: bool,
closed: bool,
idle_waits: Vec<Arc<Event>>,
}
struct Job<S> {
snapshot: S,
ticket: Option<SaveTicket>,
}
impl<S: Send + 'static> SaveWriter<S> {
pub(crate) fn new(
name: String,
save: impl Fn(&S) -> Result<(), Error> + Send + Sync + 'static,
done: impl Fn(&S, &Result<(), Error>) + Send + Sync + 'static,
) -> Self {
Self {
shared: Arc::new(Shared {
queue: Mutex::new(Queue {
jobs: VecDeque::new(),
running: false,
closed: false,
idle_waits: Vec::new(),
}),
work: Condvar::new(),
save: AssertUnwindSafe(Box::new(save)),
done: AssertUnwindSafe(Box::new(done)),
}),
thread: None,
name,
}
}
pub(crate) fn submit(&mut self, snapshot: S) -> SaveTicket {
let ticket = SaveTicket::default();
self.enqueue(Job {
snapshot,
ticket: Some(ticket.clone()),
});
ticket
}
pub(crate) fn submit_coalescing(&mut self, snapshot: S) {
{
let mut queue = lock(&self.shared.queue);
if let Some(last) = queue.jobs.back_mut().filter(|job| job.ticket.is_none()) {
last.snapshot = snapshot;
return;
}
}
self.enqueue(Job {
snapshot,
ticket: None,
});
}
pub(crate) fn idle(&self) -> SaveWait {
let event = Arc::new(Event::default());
let mut queue = lock(&self.shared.queue);
if queue.running || !queue.jobs.is_empty() {
queue.idle_waits.push(Arc::clone(&event));
} else {
drop(queue);
event.set();
}
SaveWait::new(event)
}
pub(crate) fn wait_idle(&self) {
self.idle().block();
}
pub(crate) fn put_back(&mut self, served: S) {
if let Err(error) = self.submit(served).take_outcome() {
tracing::warn!(
writer = %self.name,
%error,
"Could not put storage back to the served state as the object closed; \
the next start serves the state staged for a request that never finished"
);
}
}
fn enqueue(&mut self, job: Job<S>) {
if self.thread.is_none() {
let shared = Arc::clone(&self.shared);
match std::thread::Builder::new()
.name(self.name.clone())
.spawn(move || run(&shared))
{
Ok(handle) => self.thread = Some(AssertUnwindSafe(handle)),
Err(error) => {
tracing::warn!(%error, writer = %self.name, "Could not start a save thread");
run_job(&self.shared, job);
return;
}
}
}
lock(&self.shared.queue).jobs.push_back(job);
self.shared.work.notify_one();
}
}
impl<S: Send + 'static> Drop for SaveWriter<S> {
fn drop(&mut self) {
lock(&self.shared.queue).closed = true;
self.shared.work.notify_one();
if let Some(AssertUnwindSafe(thread)) = self.thread.take() {
let _ = thread.join();
}
}
}
fn run<S>(shared: &Shared<S>) {
loop {
let job = {
let mut queue = lock(&shared.queue);
loop {
if let Some(job) = queue.jobs.pop_front() {
queue.running = true;
break job;
}
if queue.closed {
return;
}
queue = shared
.work
.wait(queue)
.unwrap_or_else(PoisonError::into_inner);
}
};
run_job(shared, job);
let idle = {
let mut queue = lock(&shared.queue);
queue.running = false;
if queue.jobs.is_empty() {
std::mem::take(&mut queue.idle_waits)
} else {
Vec::new()
}
};
for event in idle {
event.set();
}
}
}
fn run_job<S>(shared: &Shared<S>, job: Job<S>) {
let Job { snapshot, ticket } = job;
let result = catch_unwind(AssertUnwindSafe(|| (shared.save.0)(&snapshot)))
.unwrap_or_else(|_| Err(Error::Encoding("persistence panicked during a save".into())));
let _ = catch_unwind(AssertUnwindSafe(|| (shared.done.0)(&snapshot, &result)));
drop(snapshot);
if let Some(ticket) = ticket {
ticket.complete(result);
}
}
pub(crate) fn sync_parent_dir(path: &Path) {
#[cfg(unix)]
{
let parent = match path.parent() {
Some(parent) if !parent.as_os_str().is_empty() => parent,
_ => Path::new("."),
};
let result = std::fs::File::open(parent).and_then(|directory| directory.sync_all());
if let Err(error) = result {
if directory_sync_unsupported(&error) {
tracing::debug!(directory = %parent.display(), %error, "Directory sync not supported here");
} else {
tracing::warn!(
directory = %parent.display(),
%error,
"Could not synchronize the directory after a save; the save stands"
);
}
}
}
#[cfg(not(unix))]
let _ = path;
}
#[cfg(unix)]
fn directory_sync_unsupported(error: &std::io::Error) -> bool {
error.raw_os_error().is_some_and(|code| {
[libc::EINVAL, libc::ENOTSUP, libc::EOPNOTSUPP, libc::EBADF].contains(&code)
})
}
#[doc(hidden)]
#[derive(Debug)]
pub enum StageStep {
Staged(SaveWait),
Busy(SaveWait),
Skip,
}
#[doc(hidden)]
#[derive(Debug, Clone, PartialEq)]
pub struct PendingWrite {
pub property: PropertyIdentifier,
pub array_index: Option<u32>,
pub value: PropertyValue,
}
#[doc(hidden)]
pub trait DurableWrites {
fn stage_write(
&mut self,
property: PropertyIdentifier,
array_index: Option<u32>,
value: &PropertyValue,
) -> StageStep;
fn stage_writes(&mut self, writes: &[PendingWrite]) -> StageStep {
for write in writes {
match self.stage_write(write.property, write.array_index, &write.value) {
StageStep::Skip => continue,
step => return step,
}
}
StageStep::Skip
}
fn stage_purge(&mut self) -> StageStep {
StageStep::Skip
}
fn commit_purge(&mut self) -> Result<(), Error> {
Err(Error::Protocol {
class: ErrorClass::OBJECT.to_raw() as u32,
code: ErrorCode::OPTIONAL_FUNCTIONALITY_NOT_SUPPORTED.to_raw() as u32,
})
}
fn release_staged_write(&mut self, staged: &SaveWait);
fn has_staged_write(&self) -> bool {
false
}
fn settle_forgotten_writes(&mut self) -> Option<SaveWait> {
None
}
}
#[cfg(test)]
#[path = "durable/tests.rs"]
mod tests;