maplibre_native_ffi/
completion.rs

1use std::ffi::c_void;
2use std::future::Future;
3use std::panic::{AssertUnwindSafe, catch_unwind};
4use std::pin::Pin;
5use std::sync::{Arc, Condvar, Mutex};
6use std::task::{Context, Poll, Waker};
7use std::time::Duration;
8
9use maplibre_native_ffi_sys as sys;
10
11use crate::convert::{FromNative, from_native};
12use crate::handle::lock;
13use crate::{Error, Result};
14
15/// The generations that a committed command published, which are zero unless
16/// it is a map command.
17///
18/// A command's future resolves to an [`Error`] instead when the command fails
19/// or is cancelled, carrying its status, diagnostic, and generation.
20#[derive(Debug, Clone, PartialEq, Eq)]
21pub struct CommandCompletion {
22    /// The map snapshot generation that the command published.
23    pub generation: u64,
24    /// The first map render update that includes the command; compare it with
25    /// a rendered frame's `map_update_generation`.
26    pub render_update_generation: u64,
27}
28
29struct State<T> {
30    result: Mutex<Option<Result<T>>>,
31    ready: Condvar,
32    waker: Mutex<Option<Waker>>,
33}
34
35type Converter<T> = Box<dyn FnOnce(&sys::mln_completion_result) -> Result<T> + Send>;
36
37struct Bridge<T> {
38    state: Arc<State<T>>,
39    convert: Mutex<Option<Converter<T>>>,
40}
41
42/// A one-shot native completion exposed as a Rust future.
43///
44/// The future resolves to an [`Error`] when the operation fails, which for a
45/// command means it failed or was cancelled.
46///
47/// Native has accepted the operation by the time the call that returns this
48/// future returns, and runs it whether or not anything awaits the result.
49/// Dropping the future doesn't cancel the operation; native still delivers and
50/// frees its result.
51#[must_use = "the operation is already submitted, and dropping its future ignores a failure; \
52              use the owner's `discarding()` view to submit a command without one"]
53pub struct NativeFuture<T> {
54    state: Arc<State<T>>,
55}
56
57impl<T> std::fmt::Debug for NativeFuture<T> {
58    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
59        f.debug_struct("NativeFuture")
60            .field("ready", &self.is_ready())
61            .finish()
62    }
63}
64
65impl<T> NativeFuture<T> {
66    /// Reports whether native has delivered the one terminal result.
67    pub fn is_ready(&self) -> bool {
68        lock(&self.state.result).is_some()
69    }
70
71    /// Blocks the calling thread until native delivers the result. An
72    /// operation that completes only while this thread services it, such as a
73    /// caller-driven render session's release, needs `wait_timeout` or an
74    /// executor instead.
75    pub fn wait(self) -> Result<T> {
76        let result = lock(&self.state.result);
77        self.state
78            .ready
79            .wait_while(result, |result| result.is_none())
80            .unwrap_or_else(|poisoned| poisoned.into_inner())
81            .take()
82            .expect("the wait ends once a result is present")
83    }
84
85    /// Blocks the calling thread for at most `timeout` and returns the result,
86    /// or gives the future back when the timeout ends first. This serves hosts
87    /// that must pump another service loop between waits rather than run an
88    /// async executor.
89    pub fn wait_timeout(self, timeout: Duration) -> std::result::Result<Result<T>, Self> {
90        let result = {
91            let result = lock(&self.state.result);
92            self.state
93                .ready
94                .wait_timeout_while(result, timeout, |result| result.is_none())
95                .unwrap_or_else(|poisoned| poisoned.into_inner())
96                .0
97                .take()
98        };
99        result.ok_or(self)
100    }
101}
102
103impl<T> Future for NativeFuture<T> {
104    type Output = Result<T>;
105
106    fn poll(self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Self::Output> {
107        let mut result = lock(&self.state.result);
108        if let Some(result) = result.take() {
109            return Poll::Ready(result);
110        }
111        *lock(&self.state.waker) = Some(context.waker().clone());
112        Poll::Pending
113    }
114}
115
116unsafe extern "C" fn complete<T>(
117    user_data: *mut c_void,
118    result: *const sys::mln_completion_result,
119) {
120    let _scope = crate::finalize::CallbackScope::enter();
121    if user_data.is_null() || result.is_null() {
122        return;
123    }
124    // SAFETY: native owns Bridge from successful submission through release.
125    let bridge = unsafe { &*user_data.cast::<Bridge<T>>() };
126    // SAFETY: completion result is borrowed for this callback.
127    let raw = unsafe { &*result };
128    let converted = if raw.status == sys::MLN_STATUS_OK {
129        match lock(&bridge.convert).take() {
130            Some(convert) => match catch_unwind(AssertUnwindSafe(|| convert(raw))) {
131                Ok(result) => result,
132                Err(payload) => {
133                    // A panic payload can itself panic when dropped across this C boundary.
134                    std::mem::forget(payload);
135                    Err(Error::new(
136                        crate::Status::NativeError,
137                        "completion conversion panicked",
138                    ))
139                }
140            },
141            None => Err(Error::new(
142                crate::Status::InvalidState,
143                "native completion ran more than once",
144            )),
145        }
146    } else {
147        Err(Error::Status {
148            status: crate::Status::from_raw(raw.status),
149            diagnostic: copy_diagnostic(raw),
150            generation: raw.generation,
151        })
152    };
153    *lock(&bridge.state.result) = Some(converted);
154    bridge.state.ready.notify_all();
155    let waker = lock(&bridge.state.waker).take();
156    if let Some(waker) = waker
157        && let Err(payload) = catch_unwind(AssertUnwindSafe(|| waker.wake()))
158    {
159        crate::callback::report_panic("mln_completion_callback", payload);
160    }
161}
162
163unsafe extern "C" fn release<T>(user_data: *mut c_void) {
164    let _scope = crate::finalize::CallbackScope::enter();
165    if !user_data.is_null() {
166        // SAFETY: native calls release once after accepted completion delivery.
167        let bridge = unsafe { Box::from_raw(user_data.cast::<Bridge<T>>()) };
168        if let Err(payload) = catch_unwind(AssertUnwindSafe(|| drop(bridge))) {
169            crate::callback::report_panic("mln_user_data_release", payload);
170        }
171    }
172}
173
174/// Submits one native operation whose completion carries a value or nothing.
175/// A failed completion, including a command that fails or is cancelled,
176/// resolves the future as an error.
177pub(crate) fn submit<T, S, C>(submit: S, convert: C) -> Result<NativeFuture<T>>
178where
179    T: Send + 'static,
180    S: FnOnce(*const sys::mln_completion, *mut sys::mln_diagnostic) -> sys::mln_status,
181    C: FnOnce(&sys::mln_completion_result) -> Result<T> + Send + 'static,
182{
183    let state = Arc::new(State {
184        result: Mutex::new(None),
185        ready: Condvar::new(),
186        waker: Mutex::new(None),
187    });
188    let bridge = Box::new(Bridge {
189        state: Arc::clone(&state),
190        convert: Mutex::new(Some(Box::new(convert))),
191    });
192    let bridge = Box::into_raw(bridge);
193    let completion = sys::mln_completion {
194        size: std::mem::size_of::<sys::mln_completion>() as u32,
195        callback: Some(complete::<T>),
196        user_data: bridge.cast(),
197        release_user_data: Some(release::<T>),
198        queue: sys::mln_delivery_queue(0),
199    };
200    if let Err(error) = crate::error::check(|diagnostic| submit(&completion, diagnostic)) {
201        // SAFETY: rejected submissions invoke neither callback and retain no pointer.
202        drop(unsafe { Box::from_raw(bridge) });
203        return Err(error);
204    }
205    Ok(NativeFuture { state })
206}
207
208pub(crate) fn unit(result: &sys::mln_completion_result) -> Result<()> {
209    if !result.value.is_null() || result.value_count != 0 {
210        return Err(Error::new(
211            crate::Status::NativeError,
212            "unit completion returned a value",
213        ));
214    }
215    Ok(())
216}
217
218pub(crate) fn command(result: &sys::mln_completion_result) -> Result<CommandCompletion> {
219    unit(result)?;
220    Ok(CommandCompletion {
221        generation: result.generation,
222        render_update_generation: result.render_update_generation,
223    })
224}
225
226fn copy_diagnostic(result: &sys::mln_completion_result) -> String {
227    if result.diagnostic.data.is_null() || result.diagnostic.size == 0 {
228        return String::new();
229    }
230    // SAFETY: the diagnostic is borrowed for this callback and copied here.
231    String::from_utf8_lossy(unsafe {
232        std::slice::from_raw_parts(result.diagnostic.data.cast::<u8>(), result.diagnostic.size)
233    })
234    .into_owned()
235}
236
237pub(crate) fn copy_value<T: Copy>(result: &sys::mln_completion_result) -> Result<T> {
238    if result.value.is_null() || result.value_count != 1 {
239        return Err(Error::new(
240            crate::Status::NativeError,
241            "native completion returned no value",
242        ));
243    }
244    // SAFETY: the submitting API defines value as one T for this callback.
245    Ok(unsafe { result.value.cast::<T>().read_unaligned() })
246}
247
248pub(crate) fn copy_slice<T: Copy>(result: &sys::mln_completion_result) -> Result<Vec<T>> {
249    if result.value_count == 0 {
250        return Ok(Vec::new());
251    }
252    if result.value.is_null() {
253        return Err(Error::new(
254            crate::Status::NativeError,
255            "native completion returned a null slice",
256        ));
257    }
258    // SAFETY: the submitting API defines value_count T elements for this
259    // callback, value_size bytes apart.
260    unsafe {
261        crate::convert::strided_items(
262            result.value.cast::<T>(),
263            result.value_count,
264            result.value_size,
265        )
266    }
267}
268
269pub(crate) fn value<N: Copy, T: FromNative<N>>(result: &sys::mln_completion_result) -> Result<T> {
270    let value = copy_value::<N>(result)?;
271    // SAFETY: the value and the storage it points to are borrowed for the
272    // completion callback, which runs this converter.
273    unsafe { from_native(value) }
274}
275
276pub(crate) fn optional<N: Copy, T: FromNative<N>>(
277    result: &sys::mln_completion_result,
278) -> Result<Option<T>> {
279    if result.value_count == 0 {
280        return Ok(None);
281    }
282    value::<N, T>(result).map(Some)
283}
284
285pub(crate) fn list<N: Copy, T: FromNative<N>>(
286    result: &sys::mln_completion_result,
287) -> Result<Vec<T>> {
288    copy_slice::<N>(result)?
289        .into_iter()
290        // SAFETY: as for `value`.
291        .map(|value| unsafe { from_native(value) })
292        .collect()
293}
294
295#[cfg(test)]
296mod tests {
297    use std::sync::atomic::{AtomicUsize, Ordering};
298
299    use super::*;
300
301    /// One accepted native submission, captured so a test can deliver its
302    /// completion by hand instead of driving a live map.
303    struct AcceptedSubmission {
304        callback: sys::mln_completion_callback,
305        user_data: *mut c_void,
306        release_user_data: sys::mln_user_data_release,
307    }
308
309    fn result(status: sys::mln_status) -> sys::mln_completion_result {
310        // SAFETY: an all-zero completion result is a valid empty result.
311        let mut result = unsafe { std::mem::zeroed::<sys::mln_completion_result>() };
312        result.size = std::mem::size_of::<sys::mln_completion_result>() as u32;
313        result.status = status;
314        result
315    }
316
317    impl AcceptedSubmission {
318        fn deliver(&self, status: sys::mln_status) {
319            self.deliver_result(&result(status));
320        }
321
322        fn deliver_result(&self, result: &sys::mln_completion_result) {
323            // SAFETY: the descriptor owns user_data from acceptance through
324            // release, and this is the one completion it delivers.
325            unsafe { self.callback.unwrap()(self.user_data, result) };
326        }
327
328        fn release(&self) {
329            // SAFETY: release belongs to the same accepted descriptor and runs
330            // once.
331            unsafe { self.release_user_data.unwrap()(self.user_data) };
332        }
333    }
334
335    /// Accepts a submission without calling native, returning the future and
336    /// the descriptor native would own.
337    fn accept<T, C>(convert: C) -> (NativeFuture<T>, AcceptedSubmission)
338    where
339        T: Send + 'static,
340        C: FnOnce(&sys::mln_completion_result) -> Result<T> + Send + 'static,
341    {
342        let mut accepted = None;
343        let future = submit(
344            |completion, _| {
345                // SAFETY: submit borrows a complete descriptor for this call.
346                let completion = unsafe { &*completion };
347                accepted = Some(AcceptedSubmission {
348                    callback: completion.callback,
349                    user_data: completion.user_data,
350                    release_user_data: completion.release_user_data,
351                });
352                sys::MLN_STATUS_OK
353            },
354            convert,
355        )
356        .unwrap();
357        (future, accepted.unwrap())
358    }
359
360    fn completed<T>(future: NativeFuture<T>) -> Result<T> {
361        future
362            .wait_timeout(Duration::ZERO)
363            .expect("the future has completed")
364    }
365
366    #[test]
367    fn polling_a_future_wakes_the_task_that_registered_a_waker() {
368        struct CountingWaker(AtomicUsize);
369
370        impl std::task::Wake for CountingWaker {
371            fn wake(self: Arc<Self>) {
372                self.0.fetch_add(1, Ordering::Release);
373            }
374        }
375
376        let (future, accepted) = accept(unit);
377        let waker = Arc::new(CountingWaker(AtomicUsize::new(0)));
378        let context_waker = Waker::from(Arc::clone(&waker));
379        let mut context = Context::from_waker(&context_waker);
380        let mut future = Box::pin(future);
381
382        assert!(future.as_mut().poll(&mut context).is_pending());
383        assert_eq!(waker.0.load(Ordering::Acquire), 0);
384
385        accepted.deliver(sys::MLN_STATUS_OK);
386        assert_eq!(waker.0.load(Ordering::Acquire), 1);
387        assert!(matches!(
388            future.as_mut().poll(&mut context),
389            Poll::Ready(Ok(()))
390        ));
391        accepted.release();
392    }
393
394    /// A native build whose element grew reports a value_size wider than this
395    /// binding's element, so the copy steps by it. A stride narrower than the
396    /// element cannot hold one, so the copy fails.
397    #[test]
398    fn an_array_result_is_read_at_its_value_size() {
399        // Two coordinates, each followed by a member this binding does not know.
400        #[repr(C)]
401        #[derive(Clone, Copy)]
402        struct Wide {
403            point: sys::mln_lat_lng,
404            newer: f64,
405        }
406        let wide = [(1.0, 2.0), (3.0, 4.0)].map(|(latitude, longitude)| Wide {
407            point: sys::mln_lat_lng {
408                latitude,
409                longitude,
410            },
411            newer: -1.0,
412        });
413        // SAFETY: an all-zero completion result is a valid empty result.
414        let mut result = unsafe { std::mem::zeroed::<sys::mln_completion_result>() };
415        result.status = sys::MLN_STATUS_OK;
416        result.value = wide.as_ptr().cast();
417        result.value_count = wide.len();
418        result.value_size = std::mem::size_of::<Wide>() as u32;
419
420        let points = copy_slice::<sys::mln_lat_lng>(&result).unwrap();
421        assert_eq!(
422            points
423                .iter()
424                .map(|point| (point.latitude, point.longitude))
425                .collect::<Vec<_>>(),
426            [(1.0, 2.0), (3.0, 4.0)]
427        );
428
429        result.value_size = std::mem::size_of::<sys::mln_lat_lng>() as u32 - 1;
430        assert_eq!(
431            copy_slice::<sys::mln_lat_lng>(&result)
432                .unwrap_err()
433                .status(),
434            crate::Status::NativeError
435        );
436    }
437
438    #[test]
439    fn a_panicking_converter_completes_with_an_error_before_native_release() {
440        let (future, accepted) = accept(|_| -> Result<()> { panic!("converter panic") });
441        accepted.deliver(sys::MLN_STATUS_OK);
442        assert_eq!(
443            completed(future).unwrap_err().status(),
444            crate::Status::NativeError
445        );
446        accepted.release();
447    }
448
449    #[test]
450    fn a_wait_gives_the_future_back_until_native_completes_it() {
451        let (future, accepted) = accept(unit);
452
453        let future = future.wait_timeout(Duration::ZERO).unwrap_err();
454
455        accepted.deliver(sys::MLN_STATUS_OK);
456        completed(future).unwrap();
457        accepted.release();
458    }
459
460    #[test]
461    fn a_failed_or_cancelled_command_fails_its_wait_with_its_status_diagnostic_and_generation() {
462        let diagnostic = b"style parse failed";
463        let mut failed = result(sys::MLN_STATUS_INVALID_ARGUMENT);
464        failed.generation = 7;
465        failed.diagnostic = sys::mln_buffer_view {
466            data: diagnostic.as_ptr().cast(),
467            size: diagnostic.len(),
468        };
469        let cancelled = result(sys::MLN_STATUS_CANCELLED);
470        for (delivered, status, diagnostic, generation) in [
471            (
472                failed,
473                crate::Status::InvalidArgument,
474                "style parse failed",
475                7,
476            ),
477            (cancelled, crate::Status::Cancelled, "", 0),
478        ] {
479            let (future, accepted) = accept(command);
480            accepted.deliver_result(&delivered);
481            let error = completed(future).unwrap_err();
482            assert_eq!(error.status(), status);
483            assert_eq!(error.diagnostic(), diagnostic);
484            assert_eq!(error.generation(), generation);
485            accepted.release();
486        }
487
488        let mut committed = result(sys::MLN_STATUS_OK);
489        committed.generation = 8;
490        committed.render_update_generation = 3;
491        let (future, accepted) = accept(command);
492        accepted.deliver_result(&committed);
493        assert_eq!(
494            completed(future).unwrap(),
495            CommandCompletion {
496                generation: 8,
497                render_update_generation: 3,
498            }
499        );
500        accepted.release();
501    }
502
503    #[test]
504    fn an_untimed_wait_returns_once_native_completes_the_future() {
505        let (future, accepted) = accept(unit);
506
507        let waiter = std::thread::spawn(move || future.wait());
508        accepted.deliver(sys::MLN_STATUS_OK);
509
510        waiter.join().unwrap().unwrap();
511        accepted.release();
512    }
513
514    #[test]
515    fn a_wait_times_out_and_a_dropped_future_accepts_a_late_completion() {
516        let (future, accepted) = accept(unit);
517
518        let future = future.wait_timeout(Duration::ZERO).unwrap_err();
519
520        // Dropping the future is how a caller stops waiting. Native still owns
521        // the bridge, so the late completion lands in it.
522        drop(future);
523        accepted.deliver(sys::MLN_STATUS_OK);
524        accepted.release();
525    }
526
527    #[test]
528    fn a_rejected_submission_returns_its_status() {
529        let error = submit(|_, _| sys::MLN_STATUS_INVALID_STATE, unit).unwrap_err();
530        assert_eq!(error.status(), crate::Status::InvalidState);
531    }
532}
maplibre_native_ffi/completion.rs

maplibre_native_ffi/
completion.rs
