not working destroy^

Author: marcin-zlakowski

cyfra-core/src/main/scala/io/computenode/cyfra/core/Allocation.scala

@@ -10,6 +10,8 @@ import izumi.reflect.Tag
 import java.nio.ByteBuffer
 
 trait Allocation:
+  def reportLayout[L <: Layout: LayoutBinding](layout: L): Unit 
+
   extension (buffer: GBinding[?])
     def read(bb: ByteBuffer, offset: Int = 0): Unit
 

cyfra-core/src/main/scala/io/computenode/cyfra/core/GBufferRegion.scala

@@ -9,9 +9,13 @@ import io.computenode.cyfra.dsl.Value.FromExpr
 import io.computenode.cyfra.dsl.binding.GBuffer
 import izumi.reflect.Tag
 
+import scala.util.chaining.given
 import java.nio.ByteBuffer
 
 sealed trait GBufferRegion[ReqAlloc <: Layout: LayoutBinding, ResAlloc <: Layout: LayoutBinding]:
+  def reqAllocBinding: LayoutBinding[ReqAlloc] = summon[LayoutBinding[ReqAlloc]]
+  def resAllocBinding: LayoutBinding[ResAlloc] = summon[LayoutBinding[ResAlloc]]
+
   def map[NewAlloc <: Layout: LayoutBinding](f: Allocation ?=> ResAlloc => NewAlloc): GBufferRegion[ReqAlloc, NewAlloc] =
     MapRegion(this, (alloc: Allocation) => (resAlloc: ResAlloc) => f(using alloc)(resAlloc))
 
@@ -31,13 +35,13 @@ object GBufferRegion:
       cyfraRuntime.withAllocation: allocation =>
 
         // noinspection ScalaRedundantCast
-        val steps: Seq[Allocation => Layout => Layout] = Seq.unfold(region: GBufferRegion[?, ?]):
-          case _: AllocRegion[?] => None
-          case MapRegion(req, f) =>
-            Some((f.asInstanceOf[Allocation => Layout => Layout], req))
+        val steps: Seq[(Allocation => Layout => Layout, LayoutBinding[Layout])] = Seq.unfold(region: GBufferRegion[?, ?]):
+          case _: AllocRegion[?]     => None
+          case m @ MapRegion(req, f) =>
+            Some(((f.asInstanceOf[Allocation => Layout => Layout], req.resAllocBinding.asInstanceOf[LayoutBinding[Layout]]), req))
 
-        val initAlloc = init(using allocation)
+        val initAlloc = init(using allocation).tap(allocation.reportLayout)
         val bodyAlloc = steps.foldLeft[Layout](initAlloc): (acc, step) =>
-          step(allocation)(acc)
+          step._1(allocation)(acc).tap(allocation.reportLayout(_)(using step._2))
 
         onDone(using allocation)(bodyAlloc.asInstanceOf[ResAlloc])

cyfra-runtime/src/main/scala/io/computenode/cyfra/runtime/ExecutionHandler.scala

@@ -74,6 +74,7 @@ class ExecutionHandler(runtime: VkCyfraRuntime, threadContext: VulkanThreadConte
     val externalBindings = getAllBindings(executeSteps).map(VkAllocation.getUnderlying)
     val deps = externalBindings.flatMap(_.execution.fold(Seq(_), _.toSeq))
     val pe = new PendingExecution(commandBuffer, deps, cleanup)
+    summon[VkAllocation].addExecution(pe)
     externalBindings.foreach(_.execution = Left(pe)) // TODO we assume all accesses are read-write
     result
 

cyfra-runtime/src/main/scala/io/computenode/cyfra/runtime/PendingExecution.scala

@@ -9,32 +9,50 @@ import org.lwjgl.vulkan.{VK13, VkCommandBuffer, VkCommandBufferSubmitInfo, VkSem
 
 import scala.collection.mutable
 
-class PendingExecution(protected val handle: VkCommandBuffer, val dependencies: Seq[PendingExecution], cleanup: () => Unit)(using Device)
-    extends VulkanObject[VkCommandBuffer]:
+class PendingExecution(protected val handle: VkCommandBuffer, val dependencies: Seq[PendingExecution], cleanup: () => Unit)(using Device):
   private val semaphore: Semaphore = Semaphore()
   private var fence: Option[Fence] = None
 
-  override protected def close(): Unit = cleanup()
-
-  private def setFence(otherFence: Fence): Unit =
-    if fence.isDefined then return
-    fence = Some(otherFence)
-    dependencies.foreach(_.setFence(otherFence))
+  def isPending: Boolean = fence.isEmpty
+  def isRunning: Boolean = fence.exists(!_.isSignaled)
+  def isFinished: Boolean = fence.exists(_.isSignaled)
+
+  def block(): Unit = fence.foreach(_.block())
+
+  private var closed = false
+  def isClosed: Boolean = closed
+  private def close(): Unit =
+    assert(!closed, "PendingExecution already closed")
+    assert(isFinished, "Cannot close a PendingExecution that is not finished")
+    cleanup()
+    closed = true
+
+  private var destroyed = false
+  def destroy(): Unit =
+    assert(!destroyed, "PendingExecution already destroyed")
+    assert(isFinished, "Cannot destroy a PendingExecution that is not finished")
+    if !closed then close()
+    semaphore.destroy()
+    fence.foreach(x => if x.isAlive then x.destroy())
+    destroyed = true
+
+  private def setFence(f: Fence): Unit = {
+    if !isPending then return
+    fence = Some(f)
+    dependencies.foreach(_.setFence(f))
+  }
 
-  private def gatherForSubmission: Seq[((VkCommandBuffer, Semaphore), Set[Semaphore])] = {
-    if fence.isDefined then return Seq.empty
+  private def gatherForSubmission: Seq[((VkCommandBuffer, Semaphore), Set[Semaphore])] =
+    if !isPending then return Seq.empty
     val mySubmission = ((handle, semaphore), dependencies.map(_.semaphore).toSet)
     dependencies.flatMap(_.gatherForSubmission).appended(mySubmission)
-  }
-
-  def block(): Unit =
-    fence match
-      case Some(f) => f.block()
-      case None    => throw new IllegalStateException("No fence set for this execution")
 
 object PendingExecution:
   def executeAll(executions: Seq[PendingExecution], queue: Queue)(using Device): Fence = pushStack: stack =>
-    val exec =
+    assert(executions.forall(_.isPending), "All executions must be pending")
+    assert(executions.nonEmpty, "At least one execution must be provided")
+
+    val exec: Seq[(Set[Semaphore], Set[(VkCommandBuffer, Semaphore)])] =
       val gathered = executions.flatMap(_.gatherForSubmission)
       val ordering = gathered.zipWithIndex.map(x => (x._1._1._1, x._2)).toMap
       gathered.toSet.groupMap(_._2)(_._1).toSeq.sortBy(x => x._2.map(_._1).map(ordering).min)
@@ -79,13 +97,13 @@ object PendingExecution:
     submitInfos.flip()
 
     val fence = Fence()
-    executions.foreach(_.setFence(fence))
     check(vkQueueSubmit2(queue.get, submitInfos, fence.get), "Failed to submit command buffer to queue")
+    executions.foreach(_.setFence(fence))
     fence
 
   def cleanupAll(executions: Seq[PendingExecution]): Unit =
     def cleanupRec(ex: PendingExecution): Unit =
-      if !ex.isAlive then return
-      ex.destroy()
+      if !ex.isClosed then return
+      ex.close()
       ex.dependencies.foreach(cleanupRec)
     executions.foreach(cleanupRec)

cyfra-runtime/src/main/scala/io/computenode/cyfra/runtime/VkAllocation.scala

@@ -29,6 +29,15 @@ import scala.util.chaining.*
 class VkAllocation(commandPool: CommandPool, executionHandler: ExecutionHandler)(using Allocator, Device) extends Allocation:
   given VkAllocation = this
 
+  override def reportLayout[L <: Layout: LayoutBinding](layout: L): Unit =
+    val executions = summon[LayoutBinding[L]]
+      .toBindings(layout)
+      .map(getUnderlying)
+      .flatMap(_.execution.fold(Seq(_), _.toSeq))
+      .filter(_.isPending)
+
+    PendingExecution.executeAll(executions, commandPool.queue)
+
   extension (buffer: GBinding[?])
     def read(bb: ByteBuffer, offset: Int = 0): Unit =
       val size = bb.remaining()
@@ -54,6 +63,7 @@ class VkAllocation(commandPool: CommandPool, executionHandler: ExecutionHandler)
             commandPool.freeCommandBuffer(cb)
             stagingBuffer.destroy()
           val pe = new PendingExecution(cb, binding.execution.fold(Seq(_), _.toSeq), cleanup)
+          addExecution(pe)
           binding.execution = Left(pe)
         case _ => throw new IllegalArgumentException(s"Tried to write to non-VkBinding $buffer")
 
@@ -89,8 +99,14 @@ class VkAllocation(commandPool: CommandPool, executionHandler: ExecutionHandler)
             case _                       => ???
           direct(bb)
 
+  private val executions = mutable.Buffer[PendingExecution]()
+
+  def addExecution(pe: PendingExecution): Unit =
+    executions += pe
+
   private val bindings = mutable.Buffer[VkUniform[?] | VkBuffer[?]]()
   private[cyfra] def close(): Unit =
+    executions.foreach(_.destroy())
     bindings.map(getUnderlying).foreach(_.buffer.destroy())
 
   private def getStagingBuffer(size: Int): Buffer.HostBuffer =

cyfra-runtime/src/main/scala/io/computenode/cyfra/runtime/VkBinding.scala

@@ -33,12 +33,15 @@ sealed abstract class VkBinding[T <: Value: {Tag, FromExpr}](val buffer: Buffer)
     */
   var execution: Either[PendingExecution, mutable.Buffer[PendingExecution]] = Right(mutable.Buffer.empty)
 
-  def materialise(queue: Queue)(using Device): Unit = execution match
-    case Left(exec) if exec.isAlive =>
-      PendingExecution.executeAll(Seq(exec), queue)
-      exec.block()
-      PendingExecution.cleanupAll(Seq(exec))
-    case _ => ()
+  def materialise(queue: Queue)(using Device): Unit =
+    val (pendingExecs, runningExecs) = execution.fold(Seq(_), _.toSeq).partition(_.isPending) // TODO better handle read only executions
+    if pendingExecs.nonEmpty then
+      val fence = PendingExecution.executeAll(pendingExecs, queue)
+      fence.block()
+      PendingExecution.cleanupAll(pendingExecs)
+      
+    runningExecs.foreach(_.block())
+    PendingExecution.cleanupAll(runningExecs)
 
 object VkBinding:
   def unapply(binding: GBinding[?]): Option[Buffer] = binding match