1use std::{marker, mem, os::fd, pin};
2
3use dope_sys::memory::Context as _;
4pub(in crate::driver) use epochs::Final;
5use finish::Finish;
6use o3::{
7 base::{
8 cell::{brand, region},
9 marker::ThreadBound,
10 },
11 buffer,
12 buffer::storage::{provided, provided::fixed},
13};
14use plans::Provided;
15use schedule::state::Core;
16use state::scope;
17
18use crate::{
19 Engine, driver,
20 driver::{
21 life::recovery::{Observe, Recover},
22 retained::flight,
23 route::lease::Registry,
24 schedule,
25 schedule::state,
26 storage,
27 storage::{resident, resident::Alloc},
28 },
29 io,
30};
31
32pub mod creation;
33pub mod domain;
34pub mod finish;
35pub(crate) mod native;
36mod plans;
37pub mod recovery;
38
39#[derive(Clone, Copy, Debug, Eq, PartialEq)]
41pub struct Busy;
42
43pub mod raw;
45
46pub trait Operation<S: storage::Plan, Q: ?Sized>: Sized {
48 type Output;
49
50 fn run<'a, 'p, 'd>(
51 self,
52 scope: Scope<'a, 'd>,
53 storage: storage::Loan<'a, 'p, 'd, S>,
54 source: &'a Q,
55 ) -> Self::Output
56 where
57 'd: 'p,
58 'p: 'a,
59 'd: 'a,
60 Self: 'd,
61 S: 'd,
62 Q: 'd;
63}
64
65pub(crate) struct Local<'d> {
68 pub(crate) driver: driver::Reference<'d>,
69 pub(crate) backend: fixed::Loan<'d, Engine>,
70 pub(in crate::driver) retry: bool,
71 pub(crate) flights: &'d mut flight::Arena,
72 pub(in crate::driver) resident: Alloc<'d>,
73 pub(in crate::driver) routes: Registry,
74 pub(in crate::driver) region: (
75 region::Token<region::Identity<'d>>,
76 region::Token<region::Identity<'d, 1>>,
77 ),
78 pub(crate) lifecycle: Lifecycle<'d>,
79}
80
81pub(crate) struct Lifecycle<'d> {
83 pub(crate) source: Option<fd::BorrowedFd<'d>>,
84 pub(crate) resuming: bool,
85 pub(crate) pending: bool,
86 pub(crate) report: &'d mut dyn FnMut(io::Reclaimed),
87 pub(in crate::driver) recover: &'d dyn recovery::Recover,
88}
89
90impl Lifecycle<'_> {
91 pub(crate) fn report(&mut self, event: io::Reclaimed) {
92 if matches!(event, io::Reclaimed::Shutdown) {
93 let owned = self.source.take().is_some();
94 if self.resuming && owned {
95 self.pending = true;
96 return;
97 }
98 }
99 (self.report)(event);
100 }
101}
102
103pub(in crate::driver) struct State<'d> {
104 local: Local<'d>,
105 work: Core<'d>,
106 token: brand::Token<brand::Identity<'d>>,
107}
108
109impl<'d> State<'d> {
110 fn new(local: Local<'d>, token: brand::Token<brand::Identity<'d>>) -> Self {
111 Self {
112 work: Core::new(local.driver),
113 local,
114 token,
115 }
116 }
117}
118
119mod epochs;
120
121#[repr(transparent)]
124pub struct Scope<'a, 'd> {
125 state: &'a mut State<'d>,
126}
127
128impl<'a, 'd> Scope<'a, 'd> {
129 fn new(state: &'a mut State<'d>) -> Self {
130 Self { state }
131 }
132
133 pub fn context(&mut self) -> driver::Context<'_, 'd> {
134 let state = &mut *self.state;
135 driver::Context::new(&mut state.local)
136 }
137
138 pub fn token(&mut self) -> &mut brand::Token<brand::Identity<'d>> {
139 &mut self.state.token
140 }
141
142 pub fn with_turn<R>(
144 &mut self,
145 run: impl for<'scope> FnOnce(
146 &'scope mut brand::Token<brand::Identity<'d>>,
147 driver::Context<'scope, 'd>,
148 scope::Scope<'scope, 'd>,
149 ) -> R,
150 ) -> R {
151 let state = &mut *self.state;
152 let ctx = driver::Context::new(&mut state.local);
153 run(&mut state.token, ctx, scope::Scope::new(&mut state.work))
154 }
155
156 pub fn with_driver<R>(
158 &mut self,
159 run: impl for<'scope> FnOnce(driver::Context<'scope, 'd>, scope::Scope<'scope, 'd>) -> R,
160 ) -> R {
161 let state = &mut *self.state;
162 let ctx = driver::Context::new(&mut state.local);
163 run(ctx, scope::Scope::new(&mut state.work))
164 }
165
166 pub fn driver_ref(&self) -> driver::Reference<'d> {
167 self.state.local.driver
168 }
169
170 pub fn try_boot(&self) -> Result<(), Busy> {
173 if self.state.local.flights.is_pending() {
174 Err(Busy)
175 } else {
176 Ok(())
177 }
178 }
179
180 pub fn quiesce(&mut self) {
184 let state = &mut *self.state;
185 let ctx = driver::Context::new(&mut state.local);
186 Final::suspend(ctx).resume()
187 }
188
189 #[doc(hidden)]
192 pub fn finish<R>(
193 &mut self,
194 forced: bool,
195 run: impl for<'finish> FnOnce(
196 &mut brand::Token<brand::Identity<'d>>,
197 &mut Finish<'finish, 'd>,
198 ) -> R,
199 ) -> R {
200 let state = &mut *self.state;
201 let ctx = driver::Context::new(&mut state.local);
202 let last = Final::suspend(ctx);
203 let mut finish = Finish::new(last, forced);
204 let output = run(&mut state.token, &mut finish);
205 finish.resume();
206 output
207 }
208}
209
210impl driver::Driver {
211 fn run_scope<A, Q: ?Sized, O>(
212 &mut self,
213 arenas: A,
214 source: &domain::Source<Q>,
215 op: O,
216 report: &mut dyn Observe,
217 ) -> O::Output
218 where
219 A: buffer::raw::Arenas,
220 O: Operation<storage::Scope<A>, Q>,
221 {
222 let runtime = &mut self.runtime;
223 let (report, recover) = report.split();
224 let root = Root {
225 flights: &mut runtime.flights,
226 resident: &mut runtime.resident,
227 source,
228 report,
229 recover,
230 };
231 runtime.shared.run(
232 pin::pin!((root, arenas)),
233 runtime.shared.context().buffer().layout().0.accounts(),
234 &mut runtime.backend,
235 Provided(op),
236 )
237 }
238
239 fn scope<'b, Q: ?Sized, O>(
240 &'b mut self,
241 source: &'b domain::Source<Q>,
242 op: O,
243 report: &'b mut dyn Observe,
244 ) -> O::Output
245 where
246 O: fixed::Operation<Root<'b, Q>, driver::Buffer, driver::Shared, Terminal = crate::Engine>,
247 {
248 let runtime = &mut self.runtime;
249 let (report, recover) = report.split();
250 let root = Root {
251 flights: &mut runtime.flights,
252 resident: &mut runtime.resident,
253 source,
254 report,
255 recover,
256 };
257 runtime.shared.run(
258 pin::pin!(root),
259 runtime.shared.context().buffer().layout().0.accounts(),
260 &mut runtime.backend,
261 op,
262 )
263 }
264}
265
266const _: () = {
267 assert!(mem::size_of::<Scope<'static, 'static>>() == mem::size_of::<&mut ()>());
268 assert!(mem::align_of::<Scope<'static, 'static>>() == mem::align_of::<&mut ()>());
269};
270
271struct Root<'a, Q: ?Sized> {
272 flights: &'a mut flight::Arena,
273 resident: &'a mut resident::Arena,
274 source: &'a domain::Source<Q>,
275 report: &'a mut dyn FnMut(io::Reclaimed),
276 recover: &'a dyn Recover,
277}
278
279impl<Q: ?Sized> Root<'_, Q> {
280 fn enter<'d>(
281 &'d mut self,
282 shared: provided::Root<'d, driver::Buffer, driver::Shared, ()>,
283 token: brand::Token<brand::Identity<'d>>,
284 regions: (
285 region::Token<region::Identity<'d>>,
286 region::Token<region::Identity<'d, 1>>,
287 ),
288 backend: fixed::Loan<'d, Engine>,
289 ) -> (State<'d>, &'d Q) {
290 let alloc = self.resident.enter(®ions.0);
291 let local = Local {
292 driver: driver::Reference::new(shared),
293 backend,
294 retry: false,
295 flights: &mut *self.flights,
296 resident: alloc,
297 routes: Registry::new(),
298 region: regions,
299 lifecycle: Lifecycle {
300 source: self.source.descriptor(),
301 resuming: false,
302 pending: false,
303 report: &mut *self.report,
304 recover: self.recover,
305 },
306 };
307 let state = State::new(local, token);
308 (state, &self.source.value)
309 }
310}
311
312#[must_use]
314pub struct Install<'a, 'd> {
315 owner: marker::PhantomData<(&'a mut raw::Anchor<'a>, ThreadBound)>,
316 _driver: driver::Invariant<'d>,
317}
318
319impl<'a, 'd> Install<'a, 'd> {
320 pub fn new<T: raw::Root<'a>>(root: T, driver: driver::Reference<'d>) -> Self {
322 Self {
323 owner: root.anchor().owner(),
324 _driver: driver.brand,
325 }
326 }
327}
328
329const _: () = {
330 assert!(mem::size_of::<Install<'static, 'static>>() == 0);
331 assert!(mem::align_of::<Install<'static, 'static>>() == 1);
332};