harmos_runtime/operations/
session.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10/* === Modules === */
11
12/* === Re-Exports === */
13
14/* === Imports === */
15
16// Standard Library Imports
17use std::sync::{
18    Arc,
19    atomic::{AtomicBool, Ordering},
20};
21
22// Third Party Imports
23
24// Local Imports
25use super::{
26    Manager, OperationGeneration, OperationInstanceId, OperationStatus, Running, StreamInput, StreamItem, StreamOutput, StreamSettings,
27    stream,
28};
29use crate::operations::OperationDeclaration;
30use crate::supervision::instance::{Instance, Phase};
31use crate::{Error, Refusal, Result};
32use harmos_domain::MemorySize;
33
34/* === Constants === */
35
36/* === Macros === */
37
38/* === Type Aliases === */
39
40pub type Upload<O> = Session<O, SessionInput<<<O as OperationDeclaration>::Input as crate::operations::Direction>::Item>, ()>;
41pub type Download<O> = Session<O, (), SessionOutput<<<O as OperationDeclaration>::Output as crate::operations::Direction>::Item>>;
42pub type Exchange<O> = Session<
43    O,
44    SessionInput<<<O as OperationDeclaration>::Input as crate::operations::Direction>::Item>,
45    SessionOutput<<<O as OperationDeclaration>::Output as crate::operations::Direction>::Item>,
46>;
47
48/* === Structs === */
49
50#[must_use = "dropping a streaming session requests cancellation"]
51pub struct Session<O: OperationDeclaration, I = (), R = ()> {
52    input: I,
53    output: R,
54    completion: Completion<O>,
55}
56
57pub struct SessionInput<T> {
58    sender: Option<StreamOutput<T>>,
59    ends: Arc<Ends>,
60    control: Arc<Instance>,
61}
62
63pub struct SessionOutput<T> {
64    receiver: StreamInput<T>,
65    ends: Arc<Ends>,
66    control: Arc<Instance>,
67}
68
69#[must_use = "join completion or explicitly cancel and settle"]
70pub struct Completion<O: OperationDeclaration> {
71    instance: Running<O>,
72    ends: Arc<Ends>,
73    delivered: bool,
74}
75
76// Channel Progress Only; Execution And Cleanup Remain On Instance.
77struct Ends {
78    input: AtomicBool,
79    output: AtomicBool,
80}
81
82/* === Enums === */
83
84/* === Traits === */
85
86/* === Implementations === */
87
88impl<O: OperationDeclaration, I, R> Session<O, I, R> {
89    pub fn id(&self) -> OperationInstanceId {
90        self.completion.id()
91    }
92
93    pub fn generation(&self) -> OperationGeneration {
94        self.completion.generation()
95    }
96
97    pub fn status(&self) -> OperationStatus {
98        self.completion.status()
99    }
100
101    pub fn cancel(&self) {
102        self.completion.cancel();
103    }
104
105    pub fn split(self) -> (I, R, Completion<O>) {
106        (self.input, self.output, self.completion)
107    }
108
109    pub async fn join(self) -> Result<crate::operations::Response<O::Response>> {
110        let (input, output, completion) = self.split();
111        drop(input);
112        drop(output);
113        completion.join().await
114    }
115
116    pub async fn cancel_and_settle(self) -> Result<()> {
117        self.cancel();
118        let (input, output, completion) = self.split();
119        drop(input);
120        drop(output);
121        completion.cancel_and_settle().await
122    }
123}
124
125impl<O: OperationDeclaration, T, R> Session<O, SessionInput<T>, R> {
126    pub fn finish_input(&mut self) {
127        self.input.finish_input();
128    }
129}
130impl<O: OperationDeclaration, T: MemorySize, R> Session<O, SessionInput<T>, R> {
131    pub async fn send(&self, item: T) -> Result<()> {
132        self.input.send(item).await
133    }
134
135    pub async fn send_with(&self, bytes: usize, make: impl FnOnce() -> Result<T>) -> Result<()> {
136        self.input.send_with(bytes, make).await
137    }
138}
139impl<O: OperationDeclaration, I, T> Session<O, I, SessionOutput<T>> {
140    pub async fn next(&mut self) -> Result<Option<StreamItem<T>>> {
141        self.output.next().await
142    }
143}
144
145impl<T> SessionInput<T> {
146    pub fn finish_input(&mut self) {
147        self.ends.input.store(true, Ordering::Release);
148        self.sender.take();
149    }
150}
151impl<T: MemorySize> SessionInput<T> {
152    pub async fn send(&self, item: T) -> Result<()> {
153        self.send_with(item.memory_size(), || Ok(item)).await
154    }
155
156    pub async fn send_with(&self, bytes: usize, make: impl FnOnce() -> Result<T>) -> Result<()> {
157        let sender = self.sender.as_ref().ok_or(Error::NotReady { domain: "operation input closed" })?;
158        tokio::select! {
159            biased;
160            error = ended(&self.control) => Err(error),
161            permit = sender.reserve(bytes) => {
162                let permit = permit?;
163                let value = make()?;
164                Ok(permit.send_with(|| Ok(value))?)
165            },
166        }
167    }
168}
169impl<T> Drop for SessionInput<T> {
170    fn drop(&mut self) {
171        if !self.ends.input.load(Ordering::Acquire) {
172            cancel_unsettled(&self.control);
173        }
174    }
175}
176
177impl<T> SessionOutput<T> {
178    pub async fn next(&mut self) -> Result<Option<StreamItem<T>>> {
179        if self.ends.output.load(Ordering::Acquire) {
180            return Ok(None);
181        }
182        let mut status = self.control.settlement();
183        loop {
184            let current = status.borrow_and_update().clone();
185            if current.cancellation_requested() {
186                self.receiver.close();
187                return Err(Refusal::Cancelled.into());
188            }
189            if current.outcome().is_some() {
190                self.receiver.close();
191                let item = self.receiver.try_recv().ok();
192                if item.is_none() {
193                    self.ends.output.store(true, Ordering::Release);
194                }
195                return Ok(item);
196            }
197            tokio::select! {
198                biased;
199                item = self.receiver.recv() => {
200                    if item.is_none() { self.ends.output.store(true, Ordering::Release); }
201                    return Ok(item);
202                }
203                changed = status.changed() => {
204                    if changed.is_err() { return Err(Refusal::Unavailable.into()); }
205                }
206            }
207        }
208    }
209}
210impl<T> Drop for SessionOutput<T> {
211    fn drop(&mut self) {
212        if !self.ends.output.load(Ordering::Acquire) {
213            cancel_unsettled(&self.control);
214        }
215    }
216}
217
218impl<O: OperationDeclaration> Completion<O> {
219    pub fn id(&self) -> OperationInstanceId {
220        self.instance.id()
221    }
222
223    pub fn generation(&self) -> OperationGeneration {
224        self.instance.generation()
225    }
226
227    pub fn status(&self) -> OperationStatus {
228        self.instance.status()
229    }
230
231    pub fn cancel(&self) {
232        self.instance.cancel();
233    }
234
235    pub async fn settle(&self) -> Result<()> {
236        self.instance.settle().await
237    }
238
239    pub async fn join(mut self) -> Result<crate::operations::Response<O::Response>> {
240        if !self.ends.input.load(Ordering::Acquire) || !self.ends.output.load(Ordering::Acquire) {
241            return Err(Error::InvalidInput {
242                field: "operation stream completion",
243                reason: "finish input and drain output before joining; use cancel_and_settle to abandon",
244            });
245        }
246        let result = (&mut self.instance).await;
247        self.delivered = true;
248        result
249    }
250
251    pub async fn cancel_and_settle(mut self) -> Result<()> {
252        self.instance.cancel();
253        let result = self.instance.settle().await;
254        self.delivered = true;
255        result
256    }
257}
258impl<O: OperationDeclaration> Drop for Completion<O> {
259    fn drop(&mut self) {
260        if !self.delivered {
261            cancel_unsettled(&self.instance.control);
262        }
263    }
264}
265
266impl Manager {
267    pub fn upload<O: crate::operations::OperationDeclaration>(&self, request: O::Request, settings: StreamSettings) -> Result<Upload<O>> {
268        upload(request, settings, |request, input| self.submit_endpoints::<O, _>(request, input))
269    }
270
271    pub fn download<O: crate::operations::OperationDeclaration>(
272        &self, request: O::Request, settings: StreamSettings,
273    ) -> Result<Download<O>> {
274        download(request, settings, |request, output| self.submit_endpoints::<O, _>(request, output))
275    }
276
277    pub fn exchange<O: crate::operations::OperationDeclaration>(
278        &self, request: O::Request, input: StreamSettings, output: StreamSettings,
279    ) -> Result<Exchange<O>> {
280        exchange(request, input, output, |request, endpoints| self.submit_endpoints::<O, _>(request, endpoints))
281    }
282}
283
284impl super::Context<'_> {
285    pub fn upload<O: crate::operations::OperationDeclaration>(&self, request: O::Request, settings: StreamSettings) -> Result<Upload<O>> {
286        upload(request, settings, |request, input| self.submit_endpoints::<O, _>(request, input))
287    }
288
289    pub fn download<O: crate::operations::OperationDeclaration>(
290        &self, request: O::Request, settings: StreamSettings,
291    ) -> Result<Download<O>> {
292        download(request, settings, |request, output| self.submit_endpoints::<O, _>(request, output))
293    }
294
295    pub fn exchange<O: crate::operations::OperationDeclaration>(
296        &self, request: O::Request, input: StreamSettings, output: StreamSettings,
297    ) -> Result<Exchange<O>> {
298        exchange(request, input, output, |request, endpoints| self.submit_endpoints::<O, _>(request, endpoints))
299    }
300}
301
302#[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
303impl super::Caller {
304    pub fn upload<O: crate::operations::OperationDeclaration>(&self, request: O::Request, settings: StreamSettings) -> Result<Upload<O>> {
305        upload(request, settings, |request, input| self.submit_endpoints::<O, _>(request, input))
306    }
307
308    pub fn download<O: crate::operations::OperationDeclaration>(
309        &self, request: O::Request, settings: StreamSettings,
310    ) -> Result<Download<O>> {
311        download(request, settings, |request, output| self.submit_endpoints::<O, _>(request, output))
312    }
313
314    pub fn exchange<O: crate::operations::OperationDeclaration>(
315        &self, request: O::Request, input: StreamSettings, output: StreamSettings,
316    ) -> Result<Exchange<O>> {
317        exchange(request, input, output, |request, endpoints| self.submit_endpoints::<O, _>(request, endpoints))
318    }
319}
320
321/* === Public Functions === */
322
323/* === Private Functions === */
324
325fn upload<O: crate::operations::OperationDeclaration>(
326    request: O::Request, settings: StreamSettings,
327    submit: impl FnOnce(
328        O::Request,
329        StreamInput<<<O as OperationDeclaration>::Input as crate::operations::Direction>::Item>,
330    ) -> Result<Running<O>>,
331) -> Result<Upload<O>> {
332    let (sender, input) = stream(settings)?;
333    let instance = submit(request, input)?;
334    let ends = Arc::new(Ends {
335        input: AtomicBool::new(false),
336        output: AtomicBool::new(true),
337    });
338    Ok(Session {
339        input: SessionInput {
340            sender: Some(sender),
341            ends: ends.clone(),
342            control: instance.control.clone(),
343        },
344        output: (),
345        completion: Completion { instance, ends, delivered: false },
346    })
347}
348fn download<O: crate::operations::OperationDeclaration>(
349    request: O::Request, settings: StreamSettings,
350    submit: impl FnOnce(
351        O::Request,
352        StreamOutput<<<O as OperationDeclaration>::Output as crate::operations::Direction>::Item>,
353    ) -> Result<Running<O>>,
354) -> Result<Download<O>> {
355    let (output, receiver) = stream(settings)?;
356    let instance = submit(request, output)?;
357    let ends = Arc::new(Ends {
358        input: AtomicBool::new(true),
359        output: AtomicBool::new(false),
360    });
361    Ok(Session {
362        input: (),
363        output: SessionOutput {
364            receiver,
365            ends: ends.clone(),
366            control: instance.control.clone(),
367        },
368        completion: Completion { instance, ends, delivered: false },
369    })
370}
371fn exchange<O: crate::operations::OperationDeclaration>(
372    request: O::Request, input: StreamSettings, output: StreamSettings,
373    submit: impl FnOnce(
374        O::Request,
375        (
376            StreamInput<<<O as OperationDeclaration>::Input as crate::operations::Direction>::Item>,
377            StreamOutput<<<O as OperationDeclaration>::Output as crate::operations::Direction>::Item>,
378        ),
379    ) -> Result<Running<O>>,
380) -> Result<Exchange<O>> {
381    let (sender, input) = stream(input)?;
382    let (output, receiver) = stream(output)?;
383    let instance = submit(request, (input, output))?;
384    let ends = Arc::new(Ends {
385        input: AtomicBool::new(false),
386        output: AtomicBool::new(false),
387    });
388    Ok(Session {
389        input: SessionInput {
390            sender: Some(sender),
391            ends: ends.clone(),
392            control: instance.control.clone(),
393        },
394        output: SessionOutput {
395            receiver,
396            ends: ends.clone(),
397            control: instance.control.clone(),
398        },
399        completion: Completion { instance, ends, delivered: false },
400    })
401}
402fn cancel_unsettled(control: &Instance) {
403    if control.status().phase() != Phase::Settled {
404        control.cancel();
405    }
406}
407async fn ended(control: &Instance) -> Error {
408    let mut status = control.settlement();
409    loop {
410        let current = status.borrow_and_update().clone();
411        if current.cancellation_requested() {
412            return Refusal::Cancelled.into();
413        }
414        if current.outcome().is_some() {
415            return Error::NotReady { domain: "operation completed" };
416        }
417        if status.changed().await.is_err() {
418            return Refusal::Unavailable.into();
419        }
420    }
421}
422
423/* === Tests === */
harmos_runtime/operations/session.rs

harmos_runtime/operations/
session.rs
