wireshift_core/
completion.rs

1//! Completion types and typed request handles.
2
3use std::any::Any;
4use std::marker::PhantomData;
5use std::sync::mpsc::{Receiver, TryRecvError};
6use std::sync::Arc;
7use std::time::{Duration, Instant};
8
9use crate::error::{Error, Result};
10use crate::ring::RingInner;
11
12/// High-level completion kind.
13///
14/// ```rust
15/// use wireshift::CompletionKind;
16///
17/// assert_eq!(CompletionKind::Completed.as_str(), "completed");
18/// ```
19#[derive(Clone, Copy, Debug, Eq, PartialEq)]
20#[non_exhaustive]
21pub enum CompletionKind {
22    /// The operation completed successfully.
23    Completed,
24    /// The operation failed.
25    Failed,
26    /// The operation was canceled.
27    Canceled,
28}
29
30impl CompletionKind {
31    /// Returns a stable string representation.
32    #[must_use]
33    pub fn as_str(self) -> &'static str {
34        match self {
35            Self::Completed => "completed",
36            Self::Failed => "failed",
37            Self::Canceled => "canceled",
38        }
39    }
40}
41
42/// Metadata for one completed submission.
43///
44/// ```no_run
45/// use wireshift::{CompletionKind, Ring, RingConfig, ops::Nop};
46///
47/// let ring = Ring::new(RingConfig::default())?;
48/// let request = ring.submit(Nop)?;
49/// let event = ring.complete(None)?;
50/// assert_eq!(event.kind(), CompletionKind::Completed);
51/// # drop(request);
52/// # Ok::<(), Box<dyn std::error::Error>>(())
53/// ```
54#[derive(Clone, Debug, Eq, PartialEq)]
55pub struct CompletionEvent {
56    id: u64,
57    op_name: &'static str,
58    kind: CompletionKind,
59}
60
61impl CompletionEvent {
62    pub(crate) fn new(id: u64, op_name: &'static str, kind: CompletionKind) -> Self {
63        Self { id, op_name, kind }
64    }
65
66    /// Returns the request identifier.
67    #[must_use]
68    pub fn id(&self) -> u64 {
69        self.id
70    }
71
72    /// Returns the operation name.
73    #[must_use]
74    pub fn op_name(&self) -> &'static str {
75        self.op_name
76    }
77
78    /// Returns the completion kind.
79    #[must_use]
80    pub fn kind(&self) -> CompletionKind {
81        self.kind
82    }
83}
84
85/// Typed handle for a submitted operation.
86///
87/// ```no_run
88/// use wireshift::{Ring, RingConfig, ops::Nop};
89///
90/// let ring = Ring::new(RingConfig::default())?;
91/// let request = ring.submit(Nop)?;
92/// request.wait(None)?;
93/// # Ok::<(), Box<dyn std::error::Error>>(())
94/// ```
95pub struct Request<T> {
96    pub(crate) id: u64,
97    pub(crate) ring: Arc<RingInner>,
98    pub(crate) receiver: Receiver<Result<Box<dyn Any + Send>>>,
99    pub(crate) marker: PhantomData<T>,
100}
101
102impl<T: Send + 'static> Request<T> {
103    /// Returns the stable request identifier.
104    #[must_use]
105    pub fn id(&self) -> u64 {
106        self.id
107    }
108
109    /// Waits until the operation completes or the optional timeout expires.
110    pub fn wait(self, timeout: Option<Duration>) -> Result<T> {
111        let deadline = timeout
112            .or(self.ring.config.default_timeout)
113            .map(|duration| Instant::now() + duration);
114        loop {
115            match self.receiver.try_recv() {
116                Ok(result) => {
117                    let boxed = result?;
118                    return boxed.downcast::<T>().map(|value| *value).map_err(|_| {
119                        Error::completion(
120                            "typed completion downcast failed",
121                            "ensure the request is awaited with the same output type it was submitted with",
122                        )
123                    });
124                }
125                Err(TryRecvError::Disconnected) => {
126                    return Err(Error::completion(
127                        "request completion channel disconnected",
128                        "keep the ring alive until all in-flight requests are completed",
129                    ));
130                }
131                Err(TryRecvError::Empty) => {}
132            }
133            let remaining = deadline.map(|target| target.saturating_duration_since(Instant::now()));
134            if matches!(remaining, Some(duration) if duration.is_zero()) {
135                return Err(Error::timeout(
136                    "request timed out while waiting for completion",
137                    "increase the timeout or reduce backend load",
138                ));
139            }
140            let step = remaining
141                .unwrap_or_else(|| Duration::from_millis(50))
142                .min(Duration::from_millis(50));
143            match self.ring.pump_completion(Some(step)) {
144                Ok(_) | Err(Error::Timeout { .. }) => {}
145                Err(error) => return Err(error),
146            }
147        }
148    }
149}
wireshift_core/completion.rs

wireshift_core/
completion.rs
