bacnet_objects/color/saving.rs
1//! How a colour object saves its Present_Value (#1526); see
2//! [`persistence`](super::persistence) for what is kept and why a change
3//! doesn't wait for its save.
4//!
5//! The saves go through a [`SaveWriter`], the writer type the durable
6//! objects share ([`crate::durable`]); each object with storage owns one,
7//! with its own thread for as long as the object lives. A save is queued
8//! under the database guard, runs on that thread, and is coalesced while one
9//! waits to start. Nothing is staged. A failed save isn't retried at once:
10//! the next change, the server's stop or the object's drop tries again. The server's `stop()` asks the object to settle
11//! (`settle_forgotten_writes`): it queues the value again if the last save
12//! failed and hands back the wait for its saves, joined to the Tags
13//! storage's wait when the object has one. Dropping the object does the
14//! same, and the writer's drop waits for those saves.
15
16use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
17use std::sync::Arc;
18
19use bacnet_types::error::Error;
20use bacnet_types::primitives::ObjectIdentifier;
21
22use super::persistence::{ColorOutputPersistence, ColorOutputSnapshot, ColorOutputValue};
23use crate::durable::{SaveWait, SaveWriter};
24
25/// A colour object's Present_Value storage, if it has one.
26///
27/// A clone, as an application clone or a COV snapshot makes, has none: the
28/// storage stays with the original, as its Tags storage does.
29#[derive(Default)]
30pub(super) struct OutputStorage(Option<OutputStore>);
31
32impl Clone for OutputStorage {
33 fn clone(&self) -> Self {
34 Self(None)
35 }
36}
37
38/// Lifetime totals of one colour object's Present_Value saves, shared with
39/// the object: take it with `output_save_counters` (on
40/// [`ColorObject`](super::ColorObject) or
41/// [`ColorTemperatureObject`](super::ColorTemperatureObject)) before the
42/// object joins a database, and read it at any time after. A failed save
43/// refuses nothing, so this is where one shows besides the log: until a
44/// retry lands, storage holds an older value than the object serves.
45#[derive(Clone, Debug, Default)]
46pub struct ColorOutputSaveCounters {
47 failed_saves: Arc<AtomicU64>,
48}
49
50impl ColorOutputSaveCounters {
51 /// Saves the persistence refused, each retry that failed included.
52 /// Saturates at `u64::MAX`.
53 pub fn failed_saves(&self) -> u64 {
54 self.failed_saves.load(Ordering::Relaxed)
55 }
56
57 fn record_failure(&self) {
58 #[allow(deprecated, reason = "try_update needs Rust 1.95; the MSRV is 1.93")]
59 let _ = self
60 .failed_saves
61 .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |n| {
62 Some(n.saturating_add(1))
63 });
64 }
65}
66
67struct OutputStore {
68 writer: SaveWriter<ColorOutputSnapshot>,
69 counters: ColorOutputSaveCounters,
70 /// The value of the last save queued, or the one loaded.
71 queued: Option<ColorOutputValue>,
72 /// The value the object holds now, as it last reported it.
73 current: ColorOutputValue,
74 /// Whether the last save the writer ran failed. Its done hook sets it,
75 /// so it tells the outcome of the latest save.
76 last_failed: Arc<AtomicBool>,
77}
78
79impl OutputStorage {
80 /// Take `persistence` for the object `oid`, which holds `current`, and
81 /// load what it saved. Returns the saved value, if any; the caller checks
82 /// it fits the object before taking it.
83 pub(super) fn open(
84 &mut self,
85 oid: ObjectIdentifier,
86 current: ColorOutputValue,
87 persistence: Arc<dyn ColorOutputPersistence>,
88 ) -> Result<Option<ColorOutputValue>, Error> {
89 let loaded = persistence.load(oid)?.map(|saved| saved.present_value);
90 let last_failed = Arc::new(AtomicBool::new(false));
91 let failed = Arc::clone(&last_failed);
92 let counters = ColorOutputSaveCounters::default();
93 let counted = counters.clone();
94 let writer = SaveWriter::new(
95 format!(
96 "bacnet-color-{}-{}-save",
97 oid.object_type().to_raw(),
98 oid.instance_number()
99 ),
100 move |snapshot: &ColorOutputSnapshot| persistence.save(oid, snapshot),
101 move |_, result| {
102 failed.store(result.is_err(), Ordering::Release);
103 if let Err(error) = result {
104 counted.record_failure();
105 tracing::warn!(
106 object = %oid,
107 %error,
108 "Failed to save a colour Present_Value; the next change or stop tries again"
109 );
110 }
111 },
112 );
113 self.0 = Some(OutputStore {
114 writer,
115 counters,
116 queued: loaded,
117 // The object takes a loaded value as Present_Value (a Color
118 // Temperature clamps it, and its next `note` saves the clamped
119 // one). `current` is only saved again after a failed save, and
120 // every save follows a `note` that sets it, so this is a
121 // starting point, not a value storage can receive unnoted.
122 current: loaded.unwrap_or(current),
123 last_failed,
124 });
125 Ok(loaded)
126 }
127
128 /// The save counters, once the object has storage.
129 pub(super) fn counters(&self) -> Option<ColorOutputSaveCounters> {
130 self.0.as_ref().map(|store| store.counters.clone())
131 }
132
133 /// Whether the object has storage.
134 pub(super) fn is_some(&self) -> bool {
135 self.0.is_some()
136 }
137
138 /// The object holds `value` now: queue a save of it when it differs from
139 /// the last one queued, or the last save failed.
140 pub(super) fn note(&mut self, value: ColorOutputValue) {
141 let Some(store) = &mut self.0 else {
142 return;
143 };
144 store.current = value;
145 if store.queued != Some(value) || store.last_failed.load(Ordering::Acquire) {
146 store.queue();
147 }
148 }
149
150 /// Settle, as the server stops: queue the value again if the last save
151 /// failed, and return the wait for every save queued so far. `None`
152 /// without storage.
153 pub(super) fn settle(&mut self) -> Option<SaveWait> {
154 let store = self.0.as_mut()?;
155 store.retry_failed();
156 Some(store.writer.idle())
157 }
158
159 /// Block until the saves queued so far have run.
160 pub(super) fn wait_for_saves(&self) {
161 if let Some(store) = &self.0 {
162 store.writer.wait_idle();
163 }
164 }
165}
166
167impl OutputStore {
168 fn queue(&mut self) {
169 self.writer.submit_coalescing(ColorOutputSnapshot {
170 present_value: self.current,
171 });
172 self.queued = Some(self.current);
173 }
174
175 fn retry_failed(&mut self) {
176 if self.last_failed.load(Ordering::Acquire) {
177 self.queue();
178 }
179 }
180}
181
182impl Drop for OutputStore {
183 /// One more try at a failed save; the writer's own drop, which follows,
184 /// waits for it.
185 fn drop(&mut self) {
186 self.retry_failed();
187 }
188}
189
190/// Implement [`DurableWrites`](crate::durable::DurableWrites) for a colour
191/// object with `profile: ProfileState` and `output: OutputStorage` fields.
192/// Tags writes stage on the profile as before; Present_Value never stages,
193/// and settling joins its saves to the profile's.
194macro_rules! impl_color_durable {
195 ($object:ty) => {
196 impl $crate::durable::DurableWrites for $object {
197 fn stage_write(
198 &mut self,
199 property: bacnet_types::enums::PropertyIdentifier,
200 array_index: Option<u32>,
201 value: &bacnet_types::primitives::PropertyValue,
202 ) -> $crate::durable::StageStep {
203 self.profile.stage_write(property, array_index, value)
204 }
205
206 fn stage_writes(
207 &mut self,
208 writes: &[$crate::durable::PendingWrite],
209 ) -> $crate::durable::StageStep {
210 self.profile.stage_writes(writes)
211 }
212
213 fn release_staged_write(&mut self, staged: &$crate::durable::SaveWait) {
214 self.profile.release_staged_write(staged);
215 }
216
217 fn has_staged_write(&self) -> bool {
218 self.profile.has_staged_write()
219 }
220
221 fn settle_forgotten_writes(&mut self) -> Option<$crate::durable::SaveWait> {
222 let tags = self.profile.settle_forgotten_writes();
223 let output = self.output.settle();
224 match (tags, output) {
225 (Some(tags), Some(output)) => Some(tags.and(output)),
226 (tags, output) => tags.or(output),
227 }
228 }
229 }
230 };
231}
232
233pub(super) use impl_color_durable;