Improved vulkan stability

build.sbt

@@ -91,6 +91,7 @@ lazy val foton = (project in file("cyfra-foton"))
 
 lazy val examples = (project in file("cyfra-examples"))
   .settings(commonSettings, runnerSettings)
+  .settings(libraryDependencies += "org.scala-lang.modules" % "scala-parallel-collections_3" % "1.2.0")
   .dependsOn(foton)
 
 lazy val vscode = (project in file("cyfra-vscode"))

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

@@ -11,9 +11,9 @@ import java.nio.ByteBuffer
 
 trait Allocation:
   extension (buffer: GBinding[?])
-    def read(bb: ByteBuffer, offset: Int = 0, size: Int = -1): Unit
+    def read(bb: ByteBuffer, offset: Int = 0): Unit
 
-    def write(bb: ByteBuffer, offset: Int = 0, size: Int = -1): Unit
+    def write(bb: ByteBuffer, offset: Int = 0): Unit
 
   extension [Params, EL <: Layout: LayoutBinding, RL <: Layout: LayoutBinding](execution: GExecution[Params, EL, RL])
     def execute(params: Params, layout: EL): RL

cyfra-examples/src/main/scala/io/computenode/cyfra/samples/TestingStuff.scala

@@ -11,6 +11,10 @@ import io.computenode.cyfra.dsl.{*, given}
 import io.computenode.cyfra.runtime.VkCyfraRuntime
 import org.lwjgl.BufferUtils
 import org.lwjgl.system.MemoryUtil
+
+import java.util.concurrent.atomic.AtomicInteger
+import scala.collection.parallel.CollectionConverters.given
+
 object TestingStuff:
 
   given GContext = GContext()
@@ -228,3 +232,47 @@ object TestingStuff:
         assert(buf.get(i) == expected(i), s"Mismatch at index $i: expected ${expected(i)}, got ${buf.get(i)}")
       }
     }
+
+  @main
+  def enduranceTest =
+    given runtime: VkCyfraRuntime = VkCyfraRuntime()
+    val bufferSize = 1280
+    val params = AddProgramParams(bufferSize, addA = 0, addB = 1)
+    val region = GBufferRegion
+      .allocate[AddProgramExecLayout]
+      .map: region =>
+        execution.execute(params, region)
+    val aInt = new AtomicInteger(0)
+    (1 to 10000).par.foreach: i =>
+      val inBuffers = List.fill(5)(BufferUtils.createIntBuffer(bufferSize))
+      val wbbList = inBuffers.map(MemoryUtil.memByteBuffer)
+      val rbbList = List.fill(5)(BufferUtils.createByteBuffer(bufferSize * 4))
+
+      val inData = (0 until bufferSize).toArray
+      inBuffers.foreach(_.put(inData).flip())
+      region.runUnsafe(
+        init = AddProgramExecLayout(
+          in1 = GBuffer[Int32](wbbList(0)),
+          in2 = GBuffer[Int32](wbbList(1)),
+          in3 = GBuffer[Int32](wbbList(2)),
+          in4 = GBuffer[Int32](wbbList(3)),
+          in5 = GBuffer[Int32](wbbList(4)),
+          out1 = GBuffer[Int32](bufferSize),
+          out2 = GBuffer[Int32](bufferSize),
+          out3 = GBuffer[Int32](bufferSize),
+          out4 = GBuffer[Int32](bufferSize),
+          out5 = GBuffer[Int32](bufferSize),
+        ),
+        onDone = layout => {
+          layout.out1.read(rbbList(0))
+          layout.out2.read(rbbList(1))
+          layout.out3.read(rbbList(2))
+          layout.out4.read(rbbList(3))
+          layout.out5.read(rbbList(4))
+        },
+      )
+      val prev = aInt.getAndAdd(1)
+      if prev % 100 == 0 then println(s"Iteration $prev completed")
+
+    runtime.close()
+    println("Endurance test completed successfully")

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

@@ -20,10 +20,11 @@ import io.computenode.cyfra.runtime.ExecutionHandler.{
 import io.computenode.cyfra.runtime.ExecutionHandler.DispatchType.*
 import io.computenode.cyfra.runtime.ExecutionHandler.ExecutionBinding.{BufferBinding, UniformBinding}
 import io.computenode.cyfra.utility.Utility.timed
+import io.computenode.cyfra.vulkan.{VulkanContext, VulkanThreadContext}
 import io.computenode.cyfra.vulkan.command.{CommandPool, Fence}
 import io.computenode.cyfra.vulkan.compute.ComputePipeline
 import io.computenode.cyfra.vulkan.core.Queue
-import io.computenode.cyfra.vulkan.memory.{DescriptorPool, DescriptorSet}
+import io.computenode.cyfra.vulkan.memory.{DescriptorPool, DescriptorPoolManager, DescriptorSet, DescriptorSetManager}
 import io.computenode.cyfra.vulkan.util.Util.{check, pushStack}
 import izumi.reflect.Tag
 import org.lwjgl.vulkan.VK10.*
@@ -32,25 +33,26 @@ import org.lwjgl.vulkan.{VkCommandBuffer, VkCommandBufferBeginInfo, VkDependency
 
 import scala.collection.mutable
 
-class ExecutionHandler(runtime: VkCyfraRuntime):
-  private val context = runtime.context
+class ExecutionHandler(runtime: VkCyfraRuntime, threadContext: VulkanThreadContext, context: VulkanContext):
   import context.given
 
-  private val queue: Queue = context.computeQueue // TODO multiple queues - multithreading support
-  private val descriptorPool: DescriptorPool = context.descriptorPool // TODO descriptor pool manager - descriptor allocation and reclamation support
-  private val commandPool: CommandPool = context.commandPool // TODO multiple command pools - different command pools for different workloads
+  private val dsManager: DescriptorSetManager = threadContext.descriptorSetManager
+  private val commandPool: CommandPool = threadContext.commandPool
 
   def handle[Params, EL <: Layout: LayoutBinding, RL <: Layout: LayoutBinding](execution: GExecution[Params, EL, RL], params: Params, layout: EL)(
     using VkAllocation,
   ): RL =
     val (result, shaderCalls) = interpret(execution, params, layout)
 
-    val descriptorSets = shaderCalls.map { case ShaderCall(pipeline, layout, _) =>
-      pipeline.pipelineLayout.sets.map(descriptorPool.allocate).zip(layout).map { case (set, bindings) =>
-        set.update(bindings.map(x => VkAllocation.getUnderlying(x.binding)))
-        set
-      }
-    }
+    val descriptorSets = shaderCalls.map:
+      case ShaderCall(pipeline, layout, _) =>
+        pipeline.pipelineLayout.sets
+          .map(dsManager.allocate)
+          .zip(layout)
+          .map:
+            case (set, bindings) =>
+              set.update(bindings.map(x => VkAllocation.getUnderlying(x.binding)))
+              set
 
     val dispatches: Seq[Dispatch] = shaderCalls
       .zip(descriptorSets)
@@ -74,9 +76,10 @@ class ExecutionHandler(runtime: VkCyfraRuntime):
 
       val fence = new Fence()
       timed("Vulkan render command"):
-        check(vkQueueSubmit(queue.get, submitInfo, fence.get), "Failed to submit command buffer to queue")
+        check(vkQueueSubmit(commandPool.queue.get, submitInfo, fence.get), "Failed to submit command buffer to queue")
         fence.block().destroy()
     commandPool.freeCommandBuffer(commandBuffer)
+    descriptorSets.flatten.foreach(dsManager.free)
     result
 
   private def interpret[Params, EL <: Layout: LayoutBinding, RL <: Layout: LayoutBinding](

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

@@ -28,27 +28,23 @@ class VkAllocation(commandPool: CommandPool, executionHandler: ExecutionHandler)
   given VkAllocation = this
 
   extension (buffer: GBinding[?])
-    def read(bb: ByteBuffer, offset: Int = 0, size: Int = -1): Unit =
-      val buf = getUnderlying(buffer)
-      val s = if size < 0 then buf.size - offset else size
-
-      buf match
-        case buffer: Buffer.HostBuffer   => Buffer.copyBuffer(buffer, bb, offset, 0, s)
+    def read(bb: ByteBuffer, offset: Int = 0): Unit =
+      val size = bb.remaining()
+      getUnderlying(buffer) match
+        case buffer: Buffer.HostBuffer   => buffer.copyTo(bb, offset)
         case buffer: Buffer.DeviceBuffer =>
-          val stagingBuffer = getStagingBuffer(s)
-          Buffer.copyBuffer(buffer, stagingBuffer, offset, 0, s, commandPool).block().destroy()
-          Buffer.copyBuffer(stagingBuffer, bb, 0, 0, s)
-
-    def write(bb: ByteBuffer, offset: Int = 0, size: Int = -1): Unit =
-      val buf = getUnderlying(buffer)
-      val s = if size < 0 then bb.remaining() else size
-
-      buf match
-        case buffer: Buffer.HostBuffer   => Buffer.copyBuffer(bb, buffer, offset, 0, s)
+          val stagingBuffer = getStagingBuffer(size)
+          Buffer.copyBuffer(buffer, stagingBuffer, offset, 0, size, commandPool)
+          stagingBuffer.copyTo(bb, 0)
+
+    def write(bb: ByteBuffer, offset: Int = 0): Unit =
+      val size = bb.remaining()
+      getUnderlying(buffer) match
+        case buffer: Buffer.HostBuffer   => buffer.copyFrom(bb, offset)
         case buffer: Buffer.DeviceBuffer =>
-          val stagingBuffer = getStagingBuffer(s)
-          Buffer.copyBuffer(bb, stagingBuffer, 0, 0, s)
-          Buffer.copyBuffer(stagingBuffer, buffer, 0, offset, s, commandPool).block().destroy()
+          val stagingBuffer = getStagingBuffer(size)
+          stagingBuffer.copyFrom(bb, offset)
+          Buffer.copyBuffer(stagingBuffer, buffer, 0, offset, size, commandPool)
 
   extension (buffers: GBuffer.type)
     def apply[T <: Value: {Tag, FromExpr}](length: Int): GBuffer[T] =

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

@@ -8,19 +8,19 @@ import io.computenode.cyfra.vulkan.compute.ComputePipeline
 import scala.collection.mutable
 
 class VkCyfraRuntime extends CyfraRuntime:
-  val context = new VulkanContext()
+  private val context = new VulkanContext()
   import context.given
 
-  private val executionHandler = new ExecutionHandler(this)
-
   private val shaderCache = mutable.Map.empty[String, VkShader[?]]
   private[cyfra] def getOrLoadProgram[Params, L <: Layout: {LayoutBinding, LayoutStruct}](program: GProgram[Params, L]): VkShader[L] =
     shaderCache.getOrElseUpdate(program.cacheKey, VkShader(program)).asInstanceOf[VkShader[L]]
 
   override def withAllocation(f: Allocation => Unit): Unit =
-    val allocation = new VkAllocation(context.commandPool, executionHandler)
-    f(allocation)
-    allocation.close()
+    context.withThreadContext: threadContext =>
+      val executionHandler = new ExecutionHandler(this, threadContext, context)
+      val allocation = new VkAllocation(threadContext.commandPool, executionHandler)
+      f(allocation)
+      allocation.close()
 
   def close(): Unit =
     shaderCache.values.foreach(_.underlying.destroy())

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

@@ -23,12 +23,12 @@ object VkShader:
 
     val shaderLayout = shaderBindings(summon[LayoutStruct[L]].layoutRef)
     val sets = shaderLayout.map: set =>
-      val descriptors = set.map { case Binding(binding, op) =>
-        val kind = binding match
-          case buffer: GBuffer[?]   => BindingType.StorageBuffer
-          case uniform: GUniform[?] => BindingType.Uniform
-        DescriptorInfo(kind)
-      }
+      val descriptors = set.map:
+        case Binding(binding, op) =>
+          val kind = binding match
+            case buffer: GBuffer[?]   => BindingType.StorageBuffer
+            case uniform: GUniform[?] => BindingType.Uniform
+          DescriptorInfo(kind)
       DescriptorSetInfo(descriptors)
 
     val pipeline = ComputePipeline(code, entryPoint, LayoutInfo(sets))

cyfra-vulkan/src/main/scala/io/computenode/cyfra/vulkan/VulkanContext.scala

@@ -3,35 +3,56 @@ package io.computenode.cyfra.vulkan
 import io.computenode.cyfra.utility.Logger.logger
 import io.computenode.cyfra.vulkan.VulkanContext.ValidationLayers
 import io.computenode.cyfra.vulkan.command.CommandPool
-import io.computenode.cyfra.vulkan.core.{DebugCallback, Device, Instance, Queue}
-import io.computenode.cyfra.vulkan.memory.{Allocator, DescriptorPool}
+import io.computenode.cyfra.vulkan.core.{DebugCallback, Device, Instance, PhysicalDevice, Queue}
+import io.computenode.cyfra.vulkan.memory.{Allocator, DescriptorPool, DescriptorPoolManager, DescriptorSetManager}
 import org.lwjgl.system.Configuration
 
+import java.util.concurrent.{ArrayBlockingQueue, BlockingQueue}
+import scala.util.chaining.*
+import scala.jdk.CollectionConverters.*
+
 /** @author
   *   MarconZet Created 13.04.2020
   */
 private[cyfra] object VulkanContext:
   val ValidationLayer: String = "VK_LAYER_KHRONOS_validation"
-  val SyncLayer: String = "VK_LAYER_KHRONOS_synchronization2"
   private val ValidationLayers: Boolean = System.getProperty("io.computenode.cyfra.vulkan.validation", "false").toBoolean
   if Configuration.STACK_SIZE.get() < 100 then logger.warn(s"Small stack size. Increase with org.lwjgl.system.stackSize")
 
 private[cyfra] class VulkanContext:
-  val instance: Instance = new Instance(ValidationLayers)
-  val debugCallback: Option[DebugCallback] = if ValidationLayers then Some(new DebugCallback(instance)) else None
-  given device: Device = new Device(instance)
-  given allocator: Allocator = new Allocator(instance, device)
-  val computeQueue: Queue = new Queue(device.computeQueueFamily, 0, device)
-  val descriptorPool: DescriptorPool = new DescriptorPool()
-  val commandPool: CommandPool = new CommandPool.Standard(computeQueue)
+  private val instance: Instance = new Instance(ValidationLayers)
+  private val debugCallback: Option[DebugCallback] = if ValidationLayers then Some(new DebugCallback(instance)) else None
+  private val physicalDevice = new PhysicalDevice(instance)
+  physicalDevice.assertRequirements()
+
+  given device: Device = new Device(instance, physicalDevice)
+  given allocator: Allocator = new Allocator(instance, physicalDevice, device)
+
+  private val descriptorPoolManager = new DescriptorPoolManager()
+  private val commandPools = device.getQueues.map(new CommandPool.Transient(_))
 
   logger.debug("Vulkan context created")
-  logger.debug("Running on device: " + device.physicalDeviceName)
+  logger.debug("Running on device: " + physicalDevice.name)
+
+  private val blockingQueue: BlockingQueue[CommandPool] = new ArrayBlockingQueue[CommandPool](commandPools.length).tap(_.addAll(commandPools.asJava))
+  def withThreadContext[T](f: VulkanThreadContext => T): T =
+    assert(
+      VulkanThreadContext.guard.get() == 0,
+      "VulkanThreadContext is not thread-safe. Each thread can have only one VulkanThreadContext at a time. You cannot stack VulkanThreadContext.",
+    )
+    val commandPool = blockingQueue.take()
+    val descriptorSetManager = new DescriptorSetManager(descriptorPoolManager)
+    val threadContext = new VulkanThreadContext(commandPool, descriptorSetManager)
+    VulkanThreadContext.guard.set(threadContext.hashCode())
+    try f(threadContext)
+    finally
+      blockingQueue.put(commandPool)
+      descriptorSetManager.destroy()
+      VulkanThreadContext.guard.set(0)
 
   def destroy(): Unit =
-    commandPool.destroy()
-    descriptorPool.destroy()
-    computeQueue.destroy()
+    commandPools.foreach(_.destroy())
+    descriptorPoolManager.destroy()
     allocator.destroy()
     device.destroy()
     debugCallback.foreach(_.destroy())

cyfra-vulkan/src/main/scala/io/computenode/cyfra/vulkan/VulkanThreadContext.scala

@@ -0,0 +1,11 @@
+package io.computenode.cyfra.vulkan
+
+import io.computenode.cyfra.vulkan.command.CommandPool
+import io.computenode.cyfra.vulkan.core.Device
+import io.computenode.cyfra.vulkan.memory.{DescriptorPoolManager, DescriptorSetManager}
+
+case class VulkanThreadContext(commandPool: CommandPool, descriptorSetManager: DescriptorSetManager)
+
+object VulkanThreadContext:
+  val guard: ThreadLocal[Int] = new ThreadLocal[Int]:
+    override def initialValue(): Int = 0

cyfra-vulkan/src/main/scala/io/computenode/cyfra/vulkan/command/CommandPool.scala

@@ -9,7 +9,7 @@ import org.lwjgl.vulkan.VK10.*
 /** @author
   *   MarconZet Created 13.04.2020 Copied from Wrap
   */
-private[cyfra] abstract class CommandPool private (flags: Int, queue: Queue)(using device: Device) extends VulkanObjectHandle:
+private[cyfra] abstract class CommandPool private (flags: Int, val queue: Queue)(using device: Device) extends VulkanObjectHandle:
   protected val handle: Long = pushStack: stack =>
     val createInfo = VkCommandPoolCreateInfo
       .calloc(stack)
@@ -39,11 +39,11 @@ private[cyfra] abstract class CommandPool private (flags: Int, queue: Queue)(usi
     check(vkAllocateCommandBuffers(device.get, allocateInfo, pointerBuffer), "Failed to allocate command buffers")
     0 until n map (i => pointerBuffer.get(i)) map (new VkCommandBuffer(_, device.get))
 
-  def executeCommand(block: VkCommandBuffer => Unit): Fence =
-    pushStack: stack =>
-      val commandBuffer = beginSingleTimeCommands()
-      block(commandBuffer)
-      endSingleTimeCommands(commandBuffer)
+  def executeCommand(block: VkCommandBuffer => Unit): Unit =
+    val commandBuffer = beginSingleTimeCommands()
+    block(commandBuffer)
+    endSingleTimeCommands(commandBuffer).block().destroy()
+    freeCommandBuffer(commandBuffer)
 
   private def beginSingleTimeCommands(): VkCommandBuffer =
     pushStack: stack =>
@@ -65,7 +65,7 @@ private[cyfra] abstract class CommandPool private (flags: Int, queue: Queue)(usi
         .calloc(stack)
         .sType$Default()
         .pCommandBuffers(pointerBuffer)
-      val fence = new Fence(0, () => freeCommandBuffer(commandBuffer))
+      val fence = Fence()
       check(vkQueueSubmit(queue.get, submitInfo, fence.get), "Failed to submit single time command buffer")
       fence
 
@@ -80,7 +80,7 @@ private[cyfra] abstract class CommandPool private (flags: Int, queue: Queue)(usi
     vkDestroyCommandPool(device.get, commandPool, null)
 
 object CommandPool:
-  private[cyfra] class OneTime(queue: Queue)(using device: Device)
+  private[cyfra] class Transient(queue: Queue)(using device: Device)
       extends CommandPool(VK_COMMAND_POOL_CREATE_TRANSIENT_BIT, queue)(using device: Device) // TODO check if flags should be used differently
 
   private[cyfra] class Standard(queue: Queue)(using device: Device) extends CommandPool(0, queue)(using device: Device)

cyfra-vulkan/src/main/scala/io/computenode/cyfra/vulkan/command/Fence.scala

@@ -9,7 +9,7 @@ import org.lwjgl.vulkan.VkFenceCreateInfo
 /** @author
   *   MarconZet Created 13.04.2020
   */
-private[cyfra] class Fence(flags: Int = 0, onDestroy: () => Unit = () => ())(using device: Device) extends VulkanObjectHandle:
+private[cyfra] class Fence(flags: Int = 0)(using device: Device) extends VulkanObjectHandle:
   protected val handle: Long = pushStack(stack =>
     val fenceInfo = VkFenceCreateInfo
       .calloc(stack)
@@ -23,7 +23,6 @@ private[cyfra] class Fence(flags: Int = 0, onDestroy: () => Unit = () => ())(usi
   )
 
   override def close(): Unit =
-    onDestroy.apply()
     vkDestroyFence(device.get, handle, null)
 
   def isSignaled: Boolean =