bacnet_server/server/shutdown.rs
1use super::*;
2
3#[cfg(test)]
4#[path = "request_tasks_tests.rs"]
5mod request_tasks_tests;
6
7#[cfg(test)]
8#[path = "producer_shutdown_tests.rs"]
9mod producer_shutdown_tests;
10
11#[cfg(test)]
12#[path = "transport_shutdown_tests.rs"]
13mod transport_shutdown_tests;
14
15#[cfg(test)]
16#[path = "owned_shutdown_tests.rs"]
17mod owned_shutdown_tests;
18
19async fn stop_producer(slot: &mut Option<JoinHandle<()>>) {
20 // Borrow across the join so cancellation leaves the handle recoverable.
21 // Clear synchronously after completion: a later stop must not poll it twice.
22 if let Some(task) = slot.as_mut() {
23 task.abort();
24 let _ = task.await;
25 }
26 *slot = None;
27}
28
29impl<T: TransportPort + 'static> BACnetServer<T> {
30 /// Seal egress, join admitted work, and stop the owned transport.
31 ///
32 /// Call it before dropping the server. A server dropped without it in
33 /// async code aborts its tasks and lets the object database go on Tokio's
34 /// blocking pool once they have let go of it (#1409), so the drop returns
35 /// at once, but nothing waits for the objects' last saves. Storage may
36 /// still change after such a drop returns, as those saves and the
37 /// put-back of a staged write (#1363) land, so `stop().await` before
38 /// building another server on the same storage.
39 ///
40 /// Once its requests are joined, a write a Notification Forwarder,
41 /// Notification Class or Audit Log still holds staged for one of them is
42 /// dropped, and stop waits until every save those objects have queued
43 /// has run, so storage holds the state each object serves (#1363); see
44 /// [`DurableWrites::settle_forgotten_writes`]. That wait has no limit:
45 /// storage that stalls holds stop up, and a warning naming the objects
46 /// still saving is logged after 5 s and every 30 s after that. Stop does
47 /// not wait while the application holds the database: the objects then
48 /// settle once it lets go, and put storage back when they are dropped.
49 ///
50 /// [`DurableWrites::settle_forgotten_writes`]: bacnet_objects::durable::DurableWrites::settle_forgotten_writes
51 ///
52 /// Cancelling this waiter leaves cleanup owned by the server. A later stop
53 /// joins it; after transport cleanup begins, dropping the server lets that
54 /// cleanup finish. Local mutation and broadcasts are rejected from the first
55 /// stop poll; local reads and database inspection remain available.
56 pub async fn stop(&mut self) -> Result<(), Error> {
57 if let Some(network) = &self.network {
58 network.seal_responses();
59 }
60 self.broadcaster.seal();
61 self.request_tasks.close();
62 stop_producer(&mut self.network_number_task).await;
63 if let Some(runtime) = &self.target_audit {
64 // Only the target drain retains ingress for ACK/control progress.
65 self.notification_transactions
66 .application_sealed
67 .store(true, Ordering::Release);
68 runtime.seal();
69 runtime.batches.begin_stop();
70 for task in [
71 &self.fault_detection_task,
72 &self.event_enrollment_task,
73 &self.trend_log_task,
74 &self.schedule_tick_task,
75 &self.intrinsic_reporting_task,
76 &self.binary_lighting_operation_task,
77 &self.cov_purge_task,
78 &self.cov_revisit_task,
79 ]
80 .into_iter()
81 .flatten()
82 {
83 task.abort();
84 }
85 {
86 let mut timer = self.dcc_timer.lock().await;
87 super::dcc_timer::cancel(&mut timer).await;
88 }
89 loop {
90 let finished = runtime.batches.finished.notified();
91 tokio::pin!(finished);
92 finished.as_mut().enable();
93 if runtime.batches.stopped() {
94 break;
95 }
96 finished.await;
97 }
98 } else {
99 // Other profiles retain their sequential, cancellation-safe shutdown.
100 self.notification_transactions.close();
101 }
102 // Keep the handle in self until joined: cancellation must not detach
103 // dispatch and allow a later stop to race its join consumer.
104 if let Some(task) = self.dispatch_task.as_mut() {
105 task.abort();
106 let _ = task.await;
107 }
108 self.dispatch_task = None;
109 while let Some(result) = self.request_tasks.join_next().await {
110 super::request_tasks::RequestTasks::observe(Some(result));
111 }
112 {
113 let mut timer = self.dcc_timer.lock().await;
114 super::dcc_timer::cancel(&mut timer).await;
115 }
116 stop_producer(&mut self.fault_detection_task).await;
117 stop_producer(&mut self.event_enrollment_task).await;
118 stop_producer(&mut self.trend_log_task).await;
119 stop_producer(&mut self.schedule_tick_task).await;
120 stop_producer(&mut self.intrinsic_reporting_task).await;
121 stop_producer(&mut self.binary_lighting_operation_task).await;
122 stop_producer(&mut self.cov_purge_task).await;
123 stop_producer(&mut self.cov_revisit_task).await;
124 // No request is left to take or release a staged save, and the
125 // operation task that would drop one is gone: put storage back to
126 // what each object serves before stop returns (#1363). A
127 // `write_local` still in flight isn't joined; if it staged, it finds
128 // its stage gone and saves in place under the guard.
129 super::durable_writes::settle_forgotten(&self.db).await;
130 // Nothing can own a Command or Channel run any more. End the runs let
131 // go of while the database was busy where they stood, then any run no
132 // task took up, so none is left in progress (#1252). An application
133 // holding the database doesn't hold up stop: they end once it's free.
134 crate::command_lists::end_unowned(&self.db, self.request_tasks.take_stranded());
135 // Dispatch has relinquished the sole join-consumer role. Retain the
136 // set in self across await so a cancelled stop can finish this drain.
137 while let Some(result) = self.notification_transactions.join_next().await {
138 NotificationTransactions::observe(Some(result));
139 }
140 if let Some(runtime) = &self.target_audit {
141 runtime.uninstall(&mut *self.db.write().await);
142 }
143 self.target_audit = None;
144 if let Some(error) = &self.transport_cleanup_error {
145 return Err(Error::Encoding(error.clone()));
146 }
147 if self.transport_cleanup.is_none() {
148 if let Some(network) = self.network.take() {
149 let mut network = match Arc::try_unwrap(network) {
150 Ok(network) => network,
151 Err(network) => {
152 self.network = Some(network);
153 return Err(Error::Encoding(
154 "server network still has an internal owner".into(),
155 ));
156 }
157 };
158 // Own the complete future, including cancellation-unsafe custom
159 // transport cleanup. Drop deliberately leaves this task running.
160 self.transport_cleanup = Some(tokio::spawn(async move {
161 let result = network.stop().await;
162 (network, result)
163 }));
164 }
165 }
166 let Some(cleanup) = self.transport_cleanup.as_mut() else {
167 return Ok(());
168 };
169 let outcome = cleanup.await;
170 self.transport_cleanup = None;
171 match outcome {
172 Ok((network, result)) => {
173 if result.is_err() {
174 self.network = Some(Arc::new(network));
175 }
176 result
177 }
178 Err(error) => {
179 let message = format!("transport cleanup task failed: {error}");
180 self.transport_cleanup_error = Some(message.clone());
181 Err(Error::Encoding(message))
182 }
183 }
184 }
185}
186
187impl<T: TransportPort> Drop for BACnetServer<T> {
188 /// Seal the server and abort its tasks. In async code the object
189 /// database then drops on Tokio's blocking pool (#1409): the drop returns
190 /// before the objects' last saves land, so storage may still change after
191 /// it. Call [`stop`](BACnetServer::stop) first to wait for them.
192 fn drop(&mut self) {
193 if let Some(network) = &self.network {
194 network.seal_responses();
195 }
196 self.broadcaster.seal();
197 self.notification_transactions
198 .application_sealed
199 .store(true, Ordering::Release);
200 if let Some(runtime) = &self.target_audit {
201 runtime.seal();
202 drop(runtime.batches.close());
203 }
204 self.request_tasks.close();
205 self.notification_transactions.close();
206 let mut tasks: Vec<JoinHandle<()>> = [
207 &mut self.dispatch_task,
208 &mut self.network_number_task,
209 &mut self.cov_purge_task,
210 &mut self.fault_detection_task,
211 &mut self.event_enrollment_task,
212 &mut self.trend_log_task,
213 &mut self.schedule_tick_task,
214 &mut self.intrinsic_reporting_task,
215 &mut self.binary_lighting_operation_task,
216 &mut self.cov_revisit_task,
217 ]
218 .into_iter()
219 .filter_map(Option::take)
220 .collect();
221 if let Ok(mut timer) = self.dcc_timer.try_lock() {
222 tasks.extend(timer.take());
223 }
224 for task in &tasks {
225 task.abort();
226 }
227 self.let_database_go(tasks);
228 }
229}
230
231impl<T: TransportPort> BACnetServer<T> {
232 /// Let the object database go off the async runtime as the server is
233 /// dropped (#1409).
234 ///
235 /// Dropping the database drops its objects, and a durable object waits
236 /// for the saves it has queued, after putting storage back to the state
237 /// it serves when a write is still staged (#1363). A server dropped in
238 /// async code without `stop()` would block a Tokio worker for as long as
239 /// storage takes. So when a runtime is current, the server's handle goes
240 /// to a task that waits until `tasks`, aborted, and the server's request
241 /// and notification tasks have let go of theirs; then, if nothing else
242 /// holds the database, it is dropped on the blocking pool, as DeleteObject
243 /// drops a removed object. The server's drop returns first, so storage may
244 /// still change after it. An application still holding the database
245 /// drops the last handle itself. With no runtime current, the database
246 /// drops here, as it always has.
247 fn let_database_go(&mut self, tasks: Vec<JoinHandle<()>>) {
248 let Ok(runtime) = tokio::runtime::Handle::try_current() else {
249 return;
250 };
251 let db = std::mem::take(&mut self.db);
252 let requests = Arc::clone(&self.request_tasks);
253 let notifications = Arc::clone(&self.notification_transactions);
254 runtime.spawn(async move {
255 for task in tasks {
256 let _ = task.await;
257 }
258 while let Some(result) = requests.join_next().await {
259 super::request_tasks::RequestTasks::observe(Some(result));
260 }
261 while let Some(result) = notifications.join_next().await {
262 NotificationTransactions::observe(Some(result));
263 }
264 if Arc::strong_count(&db) == 1 {
265 drop(tokio::task::spawn_blocking(move || drop(db)));
266 }
267 });
268 }
269}