Task graph [3/10]: execution (#2548)

Cargo.lock

@@ -402,7 +402,7 @@ dependencies = [
 [[package]]
 name = "concurrent-slotmap"
 version = "0.1.0"
-source = "git+https://github.com/vulkano-rs/concurrent-slotmap?rev=a65c7642f8a647739973157d0c04d07e4474ebec#a65c7642f8a647739973157d0c04d07e4474ebec"
+source = "git+https://github.com/vulkano-rs/concurrent-slotmap?rev=bf52f0a55713bb29dde3e38bc3497b03473d1628#bf52f0a55713bb29dde3e38bc3497b03473d1628"
 dependencies = [
  "virtual-buffer",
 ]

Cargo.toml

@@ -42,7 +42,7 @@ ahash = "0.8"
 # https://github.com/KhronosGroup/Vulkan-Headers/commits/main/registry/vk.xml
 ash = "0.38.0"
 bytemuck = "1.9"
-concurrent-slotmap = { git = "https://github.com/vulkano-rs/concurrent-slotmap", rev = "a65c7642f8a647739973157d0c04d07e4474ebec" }
+concurrent-slotmap = { git = "https://github.com/vulkano-rs/concurrent-slotmap", rev = "bf52f0a55713bb29dde3e38bc3497b03473d1628" }
 core-graphics-types = "0.1"
 crossbeam-queue = "0.3"
 half = "2.0"

vulkano-taskgraph/src/graph/mod.rs

@@ -1,12 +1,16 @@
 //! The task graph data structure and associated types.
 
+pub use self::execute::{ExecuteError, ResourceMap};
 use crate::{
     resource::{AccessType, BufferRange, ImageLayoutType},
     Id, InvalidSlotError, QueueFamilyType, Task, BUFFER_TAG, IMAGE_TAG, SWAPCHAIN_TAG,
 };
 use concurrent_slotmap::{IterMut, IterUnprotected, SlotId, SlotMap};
 use smallvec::SmallVec;
-use std::{borrow::Cow, error::Error, fmt, hint, iter::FusedIterator, ops::Range, sync::Arc};
+use std::{
+    borrow::Cow, cell::RefCell, error::Error, fmt, hint, iter::FusedIterator, ops::Range, slice,
+    sync::Arc,
+};
 use vulkano::{
     buffer::{Buffer, BufferCreateInfo},
     device::{Device, DeviceOwned, Queue},
@@ -19,6 +23,8 @@ use vulkano::{
     DeviceSize,
 };
 
+mod execute;
+
 const EXCLUSIVE_BIT: u32 = 1 << 6;
 const VIRTUAL_BIT: u32 = 1 << 7;
 
@@ -647,6 +653,10 @@ impl<W: ?Sized> TaskNode<W> {
 }
 
 impl ResourceAccesses {
+    fn iter(&self) -> slice::Iter<'_, ResourceAccess> {
+        self.inner.iter()
+    }
+
     pub(crate) fn contains_buffer_access(
         &self,
         id: Id<Buffer>,
@@ -655,7 +665,7 @@ impl ResourceAccesses {
     ) -> bool {
         debug_assert!(!range.is_empty());
 
-        self.inner.iter().any(|resource_access| {
+        self.iter().any(|resource_access| {
             matches!(resource_access, ResourceAccess::Buffer(a) if a.id == id
                 && a.access_type == access_type
                 && a.range.start <= range.start
@@ -674,7 +684,7 @@ impl ResourceAccesses {
         debug_assert!(!subresource_range.mip_levels.is_empty());
         debug_assert!(!subresource_range.array_layers.is_empty());
 
-        self.inner.iter().any(|resource_access| {
+        self.iter().any(|resource_access| {
             matches!(resource_access, ResourceAccess::Image(a) if a.id == id
                 && a.access_type == access_type
                 && a.layout_type == layout_type
@@ -695,7 +705,7 @@ impl ResourceAccesses {
     ) -> bool {
         debug_assert!(!array_layers.is_empty());
 
-        self.inner.iter().any(|resource_access| {
+        self.iter().any(|resource_access| {
             matches!(resource_access, ResourceAccess::Swapchain(a) if a.id == id
                 && a.access_type == access_type
                 && a.layout_type == layout_type
@@ -814,7 +824,7 @@ impl<W: ?Sized> TaskNodeBuilder<'_, W> {
     pub unsafe fn image_access_unchecked(
         &mut self,
         id: Id<Image>,
-        mut subresource_range: ImageSubresourceRange,
+        subresource_range: ImageSubresourceRange,
         access_type: AccessType,
         mut layout_type: ImageLayoutType,
     ) -> &mut Self {
@@ -908,7 +918,7 @@ pub struct ExecutableTaskGraph<W: ?Sized> {
     submissions: Vec<Submission>,
     buffer_barriers: Vec<BufferMemoryBarrier>,
     image_barriers: Vec<ImageMemoryBarrier>,
-    semaphores: Vec<Semaphore>,
+    semaphores: RefCell<Vec<Semaphore>>,
     swapchains: SmallVec<[Id<Swapchain>; 1]>,
     present_queue: Option<Arc<Queue>>,
 }

vulkano-taskgraph/src/lib.rs

@@ -3,10 +3,8 @@
 #![forbid(unsafe_op_in_unsafe_fn)]
 
 use concurrent_slotmap::SlotId;
-use graph::ResourceAccesses;
-use resource::{
-    AccessType, BufferRange, BufferState, DeathRow, ImageState, Resources, SwapchainState,
-};
+use graph::{ResourceAccesses, ResourceMap};
+use resource::{AccessType, BufferRange, BufferState, DeathRow, ImageState, SwapchainState};
 use std::{
     any::{Any, TypeId},
     cell::Cell,
@@ -16,6 +14,7 @@ use std::{
     hash::{Hash, Hasher},
     marker::PhantomData,
     ops::{Deref, DerefMut, Range, RangeBounds},
+    sync::Arc,
     thread,
 };
 use vulkano::{
@@ -117,10 +116,10 @@ impl<W: ?Sized> fmt::Debug for dyn Task<World = W> {
 ///
 /// This gives you access to the current command buffer, resources, as well as resource cleanup.
 pub struct TaskContext<'a> {
-    resources: &'a Resources,
+    resource_map: &'a ResourceMap<'a>,
     death_row: Cell<Option<&'a mut DeathRow>>,
     current_command_buffer: Cell<Option<&'a mut RawRecordingCommandBuffer>>,
-    command_buffers: Cell<Option<&'a mut Vec<RawCommandBuffer>>>,
+    command_buffers: Cell<Option<&'a mut Vec<Arc<RawCommandBuffer>>>>,
     accesses: &'a ResourceAccesses,
 }
 
@@ -160,7 +159,7 @@ impl<'a> TaskContext<'a> {
     ///
     /// [`raw_command_buffer`]: Self::raw_command_buffer
     #[inline]
-    pub unsafe fn push_command_buffer(&self, command_buffer: RawCommandBuffer) {
+    pub unsafe fn push_command_buffer(&self, command_buffer: Arc<RawCommandBuffer>) {
         let vec = self.command_buffers.take().unwrap();
         vec.push(command_buffer);
         self.command_buffers.set(Some(vec));
@@ -179,7 +178,7 @@ impl<'a> TaskContext<'a> {
     #[inline]
     pub unsafe fn extend_command_buffers(
         &self,
-        command_buffers: impl IntoIterator<Item = RawCommandBuffer>,
+        command_buffers: impl IntoIterator<Item = Arc<RawCommandBuffer>>,
     ) {
         let vec = self.command_buffers.take().unwrap();
         vec.extend(command_buffers);
@@ -189,28 +188,31 @@ impl<'a> TaskContext<'a> {
     /// Returns the buffer corresponding to `id`, or returns an error if it isn't present.
     #[inline]
     pub fn buffer(&self, id: Id<Buffer>) -> TaskResult<&'a BufferState> {
-        // SAFETY: Ensured by the caller of `Task::execute`.
-        Ok(unsafe { self.resources.buffer_unprotected(id) }?)
+        // SAFETY: The caller of `Task::execute` must ensure that `self.resource_map` maps the
+        // virtual IDs of the graph exhaustively.
+        Ok(unsafe { self.resource_map.buffer(id) }?)
     }
 
     /// Returns the image corresponding to `id`, or returns an error if it isn't present.
     #[inline]
     pub fn image(&self, id: Id<Image>) -> TaskResult<&'a ImageState> {
-        // SAFETY: Ensured by the caller of `Task::execute`.
-        Ok(unsafe { self.resources.image_unprotected(id) }?)
+        // SAFETY: The caller of `Task::execute` must ensure that `self.resource_map` maps the
+        // virtual IDs of the graph exhaustively.
+        Ok(unsafe { self.resource_map.image(id) }?)
     }
 
     /// Returns the swapchain corresponding to `id`, or returns an error if it isn't present.
     #[inline]
     pub fn swapchain(&self, id: Id<Swapchain>) -> TaskResult<&'a SwapchainState> {
-        // SAFETY: Ensured by the caller of `Task::execute`.
-        Ok(unsafe { self.resources.swapchain_unprotected(id) }?)
+        // SAFETY: The caller of `Task::execute` must ensure that `self.resource_map` maps the
+        // virtual IDs of the graph exhaustively.
+        Ok(unsafe { self.resource_map.swapchain(id) }?)
     }
 
-    /// Returns the `Resources` collection.
+    /// Returns the `ResourceMap`.
     #[inline]
-    pub fn resources(&self) -> &'a Resources {
-        self.resources
+    pub fn resource_map(&self) -> &'a ResourceMap<'a> {
+        self.resource_map
     }
 
     /// Tries to get read access to a portion of the buffer corresponding to `id`.
@@ -624,7 +626,7 @@ impl<'a> TaskContext<'a> {
     // FIXME: unsafe
     #[inline]
     pub unsafe fn destroy_buffer(&self, id: Id<Buffer>) -> TaskResult {
-        let state = unsafe { self.resources.remove_buffer(id) }?;
+        let state = unsafe { self.resource_map.resources().remove_buffer(id) }?;
         let death_row = self.death_row.take().unwrap();
         // FIXME:
         death_row.push(state.buffer().clone());
@@ -638,7 +640,7 @@ impl<'a> TaskContext<'a> {
     // FIXME: unsafe
     #[inline]
     pub unsafe fn destroy_image(&self, id: Id<Image>) -> TaskResult {
-        let state = unsafe { self.resources.remove_image(id) }?;
+        let state = unsafe { self.resource_map.resources().remove_image(id) }?;
         let death_row = self.death_row.take().unwrap();
         // FIXME:
         death_row.push(state.image().clone());
@@ -652,7 +654,7 @@ impl<'a> TaskContext<'a> {
     // FIXME: unsafe
     #[inline]
     pub unsafe fn destroy_swapchain(&self, id: Id<Swapchain>) -> TaskResult {
-        let state = unsafe { self.resources.remove_swapchain(id) }?;
+        let state = unsafe { self.resource_map.resources().remove_swapchain(id) }?;
         let death_row = self.death_row.take().unwrap();
         // FIXME:
         death_row.push(state.swapchain().clone());
@@ -905,6 +907,10 @@ impl<T> Id<T> {
     fn index(self) -> u32 {
         self.slot.index()
     }
+
+    fn tag(self) -> u32 {
+        self.slot.tag()
+    }
 }
 
 impl<T> Clone for Id<T> {

vulkano-taskgraph/src/resource.rs

@@ -8,6 +8,7 @@ use rangemap::RangeMap;
 use smallvec::SmallVec;
 use std::{
     any::Any,
+    cmp,
     hash::Hash,
     iter::FusedIterator,
     mem,
@@ -21,6 +22,7 @@ use std::{
 use thread_local::ThreadLocal;
 use vulkano::{
     buffer::{AllocateBufferError, Buffer, BufferCreateInfo},
+    command_buffer::allocator::StandardCommandBufferAllocator,
     device::{Device, DeviceOwned},
     image::{
         AllocateImageError, Image, ImageAspects, ImageCreateFlags, ImageCreateInfo, ImageLayout,
@@ -48,6 +50,7 @@ static REGISTERED_DEVICES: Mutex<Vec<usize>> = Mutex::new(Vec::new());
 #[derive(Debug)]
 pub struct Resources {
     memory_allocator: Arc<dyn MemoryAllocator>,
+    command_buffer_allocator: Arc<StandardCommandBufferAllocator>,
 
     global: epoch::GlobalHandle,
     locals: ThreadLocal<epoch::UniqueLocalHandle>,
@@ -140,10 +143,16 @@ impl Resources {
 
         registered_devices.push(device_addr);
 
+        let command_buffer_allocator = Arc::new(StandardCommandBufferAllocator::new(
+            device.clone(),
+            Default::default(),
+        ));
+
         let global = epoch::GlobalHandle::new();
 
         Resources {
             memory_allocator,
+            command_buffer_allocator,
             locals: ThreadLocal::new(),
             buffers: SlotMap::with_global(create_info.max_buffers, global.clone()),
             images: SlotMap::with_global(create_info.max_images, global.clone()),
@@ -520,6 +529,12 @@ impl Resources {
         unsafe { self.buffers.get_unprotected(id.slot) }.ok_or(InvalidSlotError::new(id))
     }
 
+    #[inline]
+    pub(crate) unsafe fn buffer_unchecked_unprotected(&self, id: Id<Buffer>) -> &BufferState {
+        // SAFETY: Enforced by the caller.
+        unsafe { self.buffers.index_unchecked_unprotected(id.index()) }
+    }
+
     /// Returns the image corresponding to `id`.
     #[inline]
     pub fn image(&self, id: Id<Image>) -> Result<Ref<'_, ImageState>> {
@@ -535,6 +550,12 @@ impl Resources {
         unsafe { self.images.get_unprotected(id.slot) }.ok_or(InvalidSlotError::new(id))
     }
 
+    #[inline]
+    pub(crate) unsafe fn image_unchecked_unprotected(&self, id: Id<Image>) -> &ImageState {
+        // SAFETY: Enforced by the caller.
+        unsafe { self.images.index_unchecked_unprotected(id.index()) }
+    }
+
     /// Returns the swapchain corresponding to `id`.
     #[inline]
     pub fn swapchain(&self, id: Id<Swapchain>) -> Result<Ref<'_, SwapchainState>> {
@@ -553,6 +574,15 @@ impl Resources {
         unsafe { self.swapchains.get_unprotected(id.slot) }.ok_or(InvalidSlotError::new(id))
     }
 
+    #[inline]
+    pub(crate) unsafe fn swapchain_unchecked_unprotected(
+        &self,
+        id: Id<Swapchain>,
+    ) -> &SwapchainState {
+        // SAFETY: Enforced by the caller.
+        unsafe { self.swapchains.index_unchecked_unprotected(id.index()) }
+    }
+
     /// Returns the [flight] corresponding to `id`.
     #[inline]
     pub fn flight(&self, id: Id<Flight>) -> Result<Ref<'_, Flight>> {
@@ -573,6 +603,10 @@ impl Resources {
         self.locals.get_or(|| self.global.register_local()).pin()
     }
 
+    pub(crate) fn command_buffer_allocator(&self) -> &Arc<StandardCommandBufferAllocator> {
+        &self.command_buffer_allocator
+    }
+
     pub(crate) fn try_advance_global_and_collect(&self, guard: &epoch::Guard<'_>) {
         if guard.try_advance_global() {
             self.buffers.try_collect(guard);
@@ -585,6 +619,9 @@ impl Resources {
 
 impl Drop for Resources {
     fn drop(&mut self) {
+        // FIXME:
+        let _ = unsafe { self.device().wait_idle() };
+
         let mut registered_devices = REGISTERED_DEVICES.lock();
 
         // This can't panic because there's no way to construct this type without the device's
@@ -624,6 +661,7 @@ impl BufferState {
         assert!(!range.is_empty());
 
         BufferAccesses {
+            range: range.clone(),
             overlapping: MutexGuard::leak(self.last_accesses.lock()).overlapping(range),
             // SAFETY: We locked the mutex above.
             _guard: unsafe { AccessesGuard::new(&self.last_accesses) },
@@ -701,6 +739,7 @@ impl ImageState {
             mip_levels: self.image.mip_levels(),
             array_layers: self.image.array_layers(),
             subresource_ranges,
+            range: 0..0,
             overlapping: last_accesses.overlapping(0..0),
             last_accesses,
             // SAFETY: We locked the mutex above.
@@ -809,6 +848,13 @@ impl SwapchainState {
         &self.images
     }
 
+    /// Returns the ID of the [flight] which owns this swapchain.
+    #[inline]
+    #[must_use]
+    pub fn flight_id(&self) -> Id<Flight> {
+        self.flight_id
+    }
+
     /// Returns the image index that's acquired in the current frame, or returns `None` if no image
     /// index is acquired.
     #[inline]
@@ -841,6 +887,7 @@ impl SwapchainState {
             mip_levels: 1,
             array_layers: self.swapchain.image_array_layers(),
             subresource_ranges,
+            range: 0..0,
             overlapping: last_accesses.overlapping(0..0),
             last_accesses,
             // SAFETY: We locked the mutex above.
@@ -931,6 +978,7 @@ pub type BufferRange = Range<DeviceSize>;
 ///
 /// [`accesses`]: BufferState::accesses
 pub struct BufferAccesses<'a> {
+    range: BufferRange,
     overlapping: rangemap::map::Overlapping<'a, DeviceSize, BufferAccess, Range<DeviceSize>>,
     _guard: AccessesGuard<'a, BufferAccess>,
 }
@@ -940,9 +988,12 @@ impl<'a> Iterator for BufferAccesses<'a> {
 
     #[inline]
     fn next(&mut self) -> Option<Self::Item> {
-        self.overlapping
-            .next()
-            .map(|(range, access)| (range.clone(), access))
+        self.overlapping.next().map(|(range, access)| {
+            let start = cmp::max(range.start, self.range.start);
+            let end = cmp::min(range.end, self.range.end);
+
+            (start..end, access)
+        })
     }
 }
 
@@ -957,6 +1008,7 @@ pub struct ImageAccesses<'a> {
     mip_levels: u32,
     array_layers: u32,
     subresource_ranges: SubresourceRanges,
+    range: Range<DeviceSize>,
     overlapping: rangemap::map::Overlapping<'a, DeviceSize, ImageAccess, Range<DeviceSize>>,
     last_accesses: &'a RangeMap<DeviceSize, ImageAccess>,
     _guard: AccessesGuard<'a, ImageAccess>,
@@ -969,11 +1021,14 @@ impl<'a> Iterator for ImageAccesses<'a> {
     fn next(&mut self) -> Option<Self::Item> {
         loop {
             if let Some((range, access)) = self.overlapping.next() {
+                let start = cmp::max(range.start, self.range.start);
+                let end = cmp::min(range.end, self.range.end);
                 let subresource_range =
-                    range_to_subresources(range.clone(), self.mip_levels, self.array_layers);
+                    range_to_subresources(start..end, self.mip_levels, self.array_layers);
 
                 break Some((subresource_range, access));
             } else if let Some(range) = self.subresource_ranges.next() {
+                self.range = range.clone();
                 self.overlapping = self.last_accesses.overlapping(range);
             } else {
                 break None;