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