Async operations

A generated launch returns a lazy Launch: construction submits no work. Await it with an executor, or import Operation and call .wait() to block. First-use library and pipeline preparation runs on a worker. Awaiting yields while that preparation, command capacity or GPU execution is pending.

Successful launches return arguments in the form passed. Owned and Arc read buffers can therefore travel through 'static futures; borrowed buffers remain borrowed until the operation resolves or is dropped. CPU readback is also async.

Sequences and dependencies

sequence(|batch| ...) puts pushed launches in one command buffer. Buffer accesses order dependent dispatches; independent dispatches may run concurrently. The runtime also orders conflicting operations across its queues. Device clones share that scheduling context; separate device contexts do not share resources implicitly.

The pipeline example verifies two stages against CPU results:

use std::{
    future::Future,
    pin::pin,
    sync::Arc,
    task::{Context, Poll, Wake, Waker},
};

use crate::stages;
use metal_oxide::{Device, LaunchConfig, sequence};

struct Unpark(std::thread::Thread);
impl Wake for Unpark {
    fn wake(self: Arc<Self>) {
        self.0.unpark();
    }
}

pub(super) fn block_on<F: Future>(future: F) -> F::Output {
    let waker = Waker::from(Arc::new(Unpark(std::thread::current())));
    let mut future = pin!(future);
    loop {
        match future.as_mut().poll(&mut Context::from_waker(&waker)) {
            Poll::Ready(result) => return result,
            Poll::Pending => std::thread::park(),
        }
    }
}

pub(super) async fn verify() -> Result<(), Box<dyn std::error::Error>> {
    let device = Device::system_default()?;
    for block in [128, 256] {
        for n in [0_u32, 1, 255, 256, 257, 1_000_003] {
            let values = (0..n).map(|i| (i % 1024) as f32 * 0.25).collect::<Vec<_>>();
            let input = device.buffer_from_slice(&values)?;
            let mut intermediate = device.buffer_zeroed::<f32>(n as usize)?;
            let mut output = device.buffer_zeroed::<f32>(n as usize)?;
            // SAFETY: both stages guard n, use disjoint allocations, and run in encoded order.
            sequence(|batch| unsafe {
                if block == 128 {
                    batch.push(stages::scale_128(
                        &device,
                        LaunchConfig::<128>::for_elements(n)?,
                        &input,
                        &mut intermediate,
                        n,
                    ))?;
                } else {
                    batch.push(stages::scale_256(
                        &device,
                        LaunchConfig::<256>::for_elements(n)?,
                        &input,
                        &mut intermediate,
                        n,
                    ))?;
                }
                batch.push(stages::finish(
                    &device,
                    LaunchConfig::<256>::for_elements(n)?,
                    &intermediate,
                    &mut output,
                    n,
                ))?;
                Ok(())
            })
            .await?;
            let scale = if block == 128 { 2.0 } else { 4.0 };
            for (i, &actual) in output.read().wait().iter().enumerate() {
                assert_eq!(
                    actual,
                    (values[i] * scale + 3.0) * 5.0 + 7.0,
                    "block={block}, n={n}, i={i}"
                );
            }
            println!(
                "pipeline block={block}, n={n}: verified on {}",
                device.name()
            );
        }
    }
    Ok(())
}

Its small executor is included because it is part of the tested source. An application may await the same operation using its existing async executor.

Composition and cancellation

Operation::then starts a fallible continuation after success. map transforms a successful result on the CPU; an error skips subsequent stages. shared() drives an operation once for multiple consumers, whose clones receive the same Arc of its result or error. If both futures::FutureExt and Operation are imported, use Operation::map(op, f) to disambiguate their methods.

Dropping an operation before submission releases its recording reservations and submits no work. Submitted GPU work continues, retaining its resources until completion; dropping the future does not block. Dropping a chain starts none of its later stages. Dropping every shared consumer stops driving the operation, while already submitted work completes.

A failed generated dispatch gates its dependents in the command buffer. A new batch that reads an unfinished writer’s output also carries its failure dependency. See debugging for the returned errors.