use crate::device_context::with_default_device_policy;
use crate::device_future::DeviceFuture;
use crate::device_operation::{DeviceOp, ExecutionContext, GraphNode, ReplayResource};
use crate::error::DeviceError;
use cuda_core::{sys, Device, IntoResult, Stream};
use std::collections::HashMap;
use std::future::IntoFuture;
use std::mem::MaybeUninit;
use std::sync::{Arc, Mutex, OnceLock};
const CU_STREAM_CAPTURE_MODE_RELAXED: sys::CUstreamCaptureMode = 2;
pub struct CudaGraph<T> {
stream: Arc<Stream>,
exec: Arc<GraphExecHandle>,
output: Option<T>,
}
struct GraphExecHandle {
capture_ctx: OnceLock<ExecutionContext>,
captured: OnceLock<Vec<Arc<dyn ReplayResource>>>,
pending: Mutex<Vec<ExecutionContext>>,
device: Arc<Device>,
cu_graph: sys::CUgraph,
cu_graph_exec: sys::CUgraphExec,
}
unsafe impl Send for GraphExecHandle {}
unsafe impl Sync for GraphExecHandle {}
impl GraphExecHandle {
fn instantiate(
device: Arc<Device>,
cu_graph: sys::CUgraph,
stream: &Stream,
) -> Result<Self, DeviceError> {
let cu_graph_exec = unsafe {
let mut cu_graph_exec = MaybeUninit::<sys::CUgraphExec>::uninit();
match sys::cuGraphInstantiateWithFlags(cu_graph_exec.as_mut_ptr(), cu_graph, 0).result()
{
Ok(()) => cu_graph_exec.assume_init(),
Err(e) => {
let _ = sys::cuGraphDestroy(cu_graph).result();
return Err(DeviceError::Driver(e));
}
}
};
let handle = Self {
capture_ctx: OnceLock::new(),
captured: OnceLock::new(),
pending: Mutex::new(Vec::new()),
device,
cu_graph,
cu_graph_exec,
};
unsafe { sys::cuGraphUpload(handle.cu_graph_exec, stream.cu_stream()).result()? };
Ok(handle)
}
fn adopt_capture(&self, ctx: ExecutionContext) {
let recorded = ctx.take_recorded();
let mut index: HashMap<usize, usize> = HashMap::with_capacity(recorded.len());
let mut captured: Vec<Arc<dyn ReplayResource>> = Vec::with_capacity(recorded.len());
for resource in recorded {
let (storage, write) = resource.replay_identity();
match index.get(&storage) {
Some(&i) => {
if write && !captured[i].replay_identity().1 {
captured[i] = resource;
}
}
None => {
index.insert(storage, captured.len());
captured.push(resource);
}
}
}
let _ = self.captured.set(captured);
let _ = self.capture_ctx.set(ctx);
}
}
impl Drop for GraphExecHandle {
fn drop(&mut self) {
let _ = self.device.bind_to_thread();
if !self.cu_graph_exec.is_null() {
let _ = unsafe { sys::cuGraphExecDestroy(self.cu_graph_exec).result() };
}
if !self.cu_graph.is_null() {
let _ = unsafe { sys::cuGraphDestroy(self.cu_graph).result() };
}
}
}
fn capture_on<R>(
stream: &Arc<Stream>,
record: impl FnOnce() -> Result<R, DeviceError>,
) -> Result<(R, Arc<GraphExecHandle>), DeviceError> {
let device = stream.device().clone();
device.bind_to_thread()?;
unsafe {
stream.begin_capture(CU_STREAM_CAPTURE_MODE_RELAXED)?;
}
let recorded = match std::panic::catch_unwind(std::panic::AssertUnwindSafe(record)) {
Ok(result) => result,
Err(payload) => {
if let Ok(cu_graph) = unsafe { stream.end_capture() } {
destroy_graph(cu_graph);
}
std::panic::resume_unwind(payload);
}
};
let end_result = unsafe { stream.end_capture() };
let (output, cu_graph) = match (recorded, end_result) {
(Err(err), Ok(cu_graph)) => {
destroy_graph(cu_graph);
return Err(err);
}
(Err(err), Err(_)) => return Err(err),
(Ok(_), Err(capture_err)) => return Err(DeviceError::Driver(capture_err)),
(Ok(_), Ok(cu_graph)) if cu_graph.is_null() => {
return Err(DeviceError::Internal(
"cuStreamEndCapture returned null graph".into(),
));
}
(Ok(output), Ok(cu_graph)) => (output, cu_graph),
};
let exec = GraphExecHandle::instantiate(device, cu_graph, stream)?;
unsafe { stream.synchronize() }?;
Ok((output, Arc::new(exec)))
}
fn destroy_graph(cu_graph: sys::CUgraph) {
if !cu_graph.is_null() {
let _ = unsafe { sys::cuGraphDestroy(cu_graph).result() };
}
}
impl<T: Send> CudaGraph<T> {
pub fn capture(
stream: Arc<Stream>,
op: impl DeviceOp<Output = T>,
) -> Result<Self, DeviceError> {
let _execution_lock = crate::device_operation::acquire_execution_lock()?;
let exec_ctx = ExecutionContext::for_capture(stream.clone());
let (output, exec) = capture_on(&stream, || unsafe { op.execute(&exec_ctx) })?;
exec.adopt_capture(exec_ctx);
Ok(Self {
stream,
exec,
output: Some(output),
})
}
pub fn take_output(&mut self) -> Option<T> {
self.output.take()
}
pub fn update<N>(&self, op: N) -> Result<(), DeviceError>
where
N: GraphNode + DeviceOp<Output = ()>,
{
let _execution_lock = crate::device_operation::acquire_execution_lock()?;
let ctx = ExecutionContext::new(self.stream.clone());
let result = unsafe { op.execute(&ctx) };
self.exec
.pending
.lock()
.unwrap_or_else(|e| e.into_inner())
.push(ctx);
result
}
pub fn launch(&self) -> GraphLaunch {
GraphLaunch {
exec: Arc::clone(&self.exec),
}
}
pub fn stream(&self) -> &Arc<Stream> {
&self.stream
}
}
pub struct GraphLaunch {
exec: Arc<GraphExecHandle>,
}
impl DeviceOp for GraphLaunch {
type Output = ();
unsafe fn execute(self, context: &ExecutionContext) -> Result<(), DeviceError> {
context.retain(self.exec.clone())?;
if let Some(captured) = self.exec.captured.get() {
for resource in captured {
resource.retain_for_launch(context)?;
}
}
let pending =
std::mem::take(&mut *self.exec.pending.lock().unwrap_or_else(|e| e.into_inner()));
for ctx in pending {
context.retain(ctx)?;
}
sys::cuGraphLaunch(
self.exec.cu_graph_exec,
context.get_cuda_stream().cu_stream(),
)
.result()?;
Ok(())
}
}
impl IntoFuture for GraphLaunch {
type Output = Result<(), DeviceError>;
type IntoFuture = DeviceFuture<(), GraphLaunch>;
fn into_future(self) -> Self::IntoFuture {
match with_default_device_policy(|policy| {
let stream = policy.next_stream()?;
let mut f = DeviceFuture::new();
f.device_operation = Some(self);
f.execution_context = Some(ExecutionContext::new(stream));
Ok(f)
}) {
Ok(Ok(future)) => future,
Ok(Err(e)) => DeviceFuture::failed(e),
Err(e) => DeviceFuture::failed(e),
}
}
}
pub struct Scope {
ctx: ExecutionContext,
_not_send: std::marker::PhantomData<*const ()>,
}
impl Scope {
pub fn record<T, N>(&self, op: N) -> Result<T, DeviceError>
where
T: Send,
N: GraphNode + DeviceOp<Output = T>,
{
unsafe { op.execute(&self.ctx) }
}
}
impl CudaGraph<()> {
pub fn scope<F>(stream: &Arc<Stream>, f: F) -> Result<Self, DeviceError>
where
F: FnOnce(&Scope) -> Result<(), DeviceError>,
{
let _execution_lock = crate::device_operation::acquire_execution_lock()?;
let scope = Scope {
ctx: ExecutionContext::for_capture(stream.clone()),
_not_send: std::marker::PhantomData,
};
let ((), exec) = capture_on(stream, || f(&scope))?;
exec.adopt_capture(scope.ctx);
Ok(CudaGraph {
stream: stream.clone(),
exec,
output: Some(()),
})
}
}
pub trait Module {
type Input: Send;
type Output: Send;
fn forward(&mut self, input: Self::Input) -> Result<Self::Output, DeviceError>;
}