From 3e12e97fe2e78446c14d9dc6a9c20a85dc546c6c Mon Sep 17 00:00:00 2001 From: Kbz-8 Date: Thu, 20 Aug 2026 14:11:12 +0200 Subject: [PATCH] [Phi] implementing fully async queues, fences and semaphores --- README.md | 44 +-- build.zig | 3 +- src/phi/PhiBinarySemaphore.zig | 41 ++- src/phi/PhiFence.zig | 81 ++++- src/phi/PhiQueue.zig | 646 +++++++++++++++++++++++++++++++-- src/phi/PhiTransport.zig | 37 +- src/phi/mic/CommandBuffer.c | 73 +++- src/phi/mic/CommandBuffer.h | 2 + src/phi/mic/Daemon.c | 7 + src/phi/mic/Logger.c | 12 +- src/phi/mic/Queue.c | 152 ++++++++ src/phi/mic/Queue.h | 9 + src/phi/mic/Transport.c | 7 + src/phi/mic/Transport.h | 1 + src/phi/shared/Protocol.h | 47 ++- src/vulkan/Queue.zig | 14 - 16 files changed, 1075 insertions(+), 101 deletions(-) create mode 100644 src/phi/mic/Queue.c create mode 100644 src/phi/mic/Queue.h diff --git a/README.md b/README.md index e1beaec..1d08ecd 100644 --- a/README.md +++ b/README.md @@ -264,7 +264,7 @@ Assume thou that functions lacking in this array are, for now, not intended to b | vkAllocateDescriptorSets | ⚙️ WIP | | vkAllocateMemory | ✅ Implemented | | vkBeginCommandBuffer | ✅ Implemented | -| vkBindBufferMemory | ⚙️ WIP | +| vkBindBufferMemory | ✅ Implemented | | vkBindImageMemory | ✅ Implemented | | vkCmdBeginQuery | ⚙️ WIP | | vkCmdBeginRenderPass | ⚙️ WIP | @@ -272,13 +272,13 @@ Assume thou that functions lacking in this array are, for now, not intended to b | vkCmdBindIndexBuffer | ⚙️ WIP | | vkCmdBindPipeline | ⚙️ WIP | | vkCmdBindVertexBuffers | ⚙️ WIP | -| vkCmdBlitImage | ✅ Implemented | +| vkCmdBlitImage | ⚙️ WIP | | vkCmdClearAttachments | ⚙️ WIP | | vkCmdClearColorImage | ⚙️ WIP | | vkCmdClearDepthStencilImage | ⚙️ WIP | | vkCmdCopyBuffer | ✅ Implemented | | vkCmdCopyBufferToImage | ✅ Implemented | -| vkCmdCopyImage | ⚙️ WIP | +| vkCmdCopyImage | ✅ Implemented | | vkCmdCopyImageToBuffer | ✅ Implemented | | vkCmdCopyQueryPoolResults | ⚙️ WIP | | vkCmdDispatch | ⚙️ WIP | @@ -312,7 +312,7 @@ Assume thou that functions lacking in this array are, for now, not intended to b | vkCmdUpdateBuffer | ⚙️ WIP | | vkCmdWaitEvents | ⚙️ WIP | | vkCmdWriteTimestamp | ⚙️ WIP | -| vkCreateBuffer | ⚙️ WIP | +| vkCreateBuffer | ✅ Implemented | | vkCreateBufferView | ⚙️ WIP | | vkCreateCommandPool | ✅ Implemented | | vkCreateComputePipelines | ⚙️ WIP | @@ -320,7 +320,7 @@ Assume thou that functions lacking in this array are, for now, not intended to b | vkCreateDescriptorSetLayout | ⚙️ WIP | | vkCreateDevice | ✅ Implemented | | vkCreateEvent | ⚙️ WIP | -| vkCreateFence | ⚙️ WIP | +| vkCreateFence | ✅ Implemented | | vkCreateFramebuffer | ⚙️ WIP | | vkCreateGraphicsPipelines | ⚙️ WIP | | vkCreateImage | ✅ Implemented | @@ -331,21 +331,21 @@ Assume thou that functions lacking in this array are, for now, not intended to b | vkCreateQueryPool | ⚙️ WIP | | vkCreateRenderPass | ⚙️ WIP | | vkCreateSampler | ⚙️ WIP | -| vkCreateSemaphore | ⚙️ WIP | +| vkCreateSemaphore | ✅ Implemented | | vkCreateShaderModule | ⚙️ WIP | | vkCreateSwapchainKHR | ⚙️ WIP | | vkCreateWaylandSurfaceKHR | ⚙️ WIP | | vkCreateWin32SurfaceKHR | ⚙️ WIP | | vkCreateXcbSurfaceKHR | ⚙️ WIP | | vkCreateXlibSurfaceKHR | ⚙️ WIP | -| vkDestroyBuffer | ⚙️ WIP | +| vkDestroyBuffer | ✅ Implemented | | vkDestroyBufferView | ⚙️ WIP | | vkDestroyCommandPool | ✅ Implemented | | vkDestroyDescriptorPool | ⚙️ WIP | | vkDestroyDescriptorSetLayout | ⚙️ WIP | | vkDestroyDevice | ✅ Implemented | | vkDestroyEvent | ⚙️ WIP | -| vkDestroyFence | ⚙️ WIP | +| vkDestroyFence | ✅ Implemented | | vkDestroyFramebuffer | ⚙️ WIP | | vkDestroyImage | ✅ Implemented | | vkDestroyImageView | ⚙️ WIP | @@ -356,11 +356,11 @@ Assume thou that functions lacking in this array are, for now, not intended to b | vkDestroyQueryPool | ⚙️ WIP | | vkDestroyRenderPass | ⚙️ WIP | | vkDestroySampler | ⚙️ WIP | -| vkDestroySemaphore | ⚙️ WIP | +| vkDestroySemaphore | ✅ Implemented | | vkDestroyShaderModule | ⚙️ WIP | | vkDestroySurfaceKHR | ⚙️ WIP | | vkDestroySwapchainKHR | ⚙️ WIP | -| vkDeviceWaitIdle | ⚙️ WIP | +| vkDeviceWaitIdle | ✅ Implemented | | vkEndCommandBuffer | ✅ Implemented | | vkEnumerateDeviceExtensionProperties | ⚙️ WIP | | vkEnumerateDeviceLayerProperties | ⚙️ WIP | @@ -369,7 +369,7 @@ Assume thou that functions lacking in this array are, for now, not intended to b | vkEnumeratePhysicalDeviceGroupsKHR | ⚙️ WIP | | vkEnumeratePhysicalDevices | ✅ Implemented | | vkFlushMappedMemoryRanges | ⚙️ WIP | -| vkFreeCommandBuffers | ⚙️ WIP | +| vkFreeCommandBuffers | ✅ Implemented | | vkFreeDescriptorSets | ⚙️ WIP | | vkFreeMemory | ✅ Implemented | | vkGetBufferDeviceAddress | ⚙️ WIP | @@ -380,14 +380,14 @@ Assume thou that functions lacking in this array are, for now, not intended to b | vkGetDeviceGroupPresentCapabilitiesKHR | ⚙️ WIP | | vkGetDeviceGroupSurfacePresentModesKHR | ⚙️ WIP | | vkGetDeviceMemoryCommitment | ⚙️ WIP | -| vkGetDeviceProcAddr | ⚙️ WIP | -| vkGetDeviceQueue | ⚙️ WIP | -| vkGetEventStatus | ⚙️ WIP | -| vkGetFenceStatus | ⚙️ WIP | +| vkGetDeviceProcAddr | ✅ Implemented | +| vkGetDeviceQueue | ✅ Implemented | +| vkGetEventStatus | ✅ Implemented | +| vkGetFenceStatus | ✅ Implemented | | vkGetImageMemoryRequirements | ✅ Implemented | | vkGetImageSparseMemoryRequirements | ⚙️ WIP | | vkGetImageSubresourceLayout | ✅ Implemented | -| vkGetInstanceProcAddr | ⚙️ WIP | +| vkGetInstanceProcAddr | ✅ Implemented | | vkGetPhysicalDeviceFeatures | ✅ Implemented | | vkGetPhysicalDeviceFormatProperties | ⚙️ WIP | | vkGetPhysicalDeviceImageFormatProperties | ✅ Implemented | @@ -408,22 +408,22 @@ Assume thou that functions lacking in this array are, for now, not intended to b | vkGetRenderAreaGranularity | ⚙️ WIP | | vkGetSwapchainImagesKHR | ⚙️ WIP | | vkInvalidateMappedMemoryRanges | ⚙️ WIP | -| vkMapMemory | ⚙️ WIP | +| vkMapMemory | ✅ Implemented | | vkMergePipelineCaches | ⚙️ WIP | | vkQueueBindSparse | ⚙️ WIP | | vkQueuePresentKHR | ⚙️ WIP | | vkQueueSubmit | ✅ Implemented | -| vkQueueWaitIdle | ⚙️ WIP | -| vkResetCommandBuffer | ⚙️ WIP | +| vkQueueWaitIdle | ✅ Implemented | +| vkResetCommandBuffer | ✅ Implemented | | vkResetCommandPool | ✅ Implemented | | vkResetDescriptorPool | ⚙️ WIP | | vkResetEvent | ⚙️ WIP | -| vkResetFences | ⚙️ WIP | +| vkResetFences | ✅ Implemented | | vkResetQueryPool | ⚙️ WIP | | vkSetEvent | ⚙️ WIP | -| vkUnmapMemory | ⚙️ WIP | +| vkUnmapMemory | ✅ Implemented | | vkUpdateDescriptorSets | ⚙️ WIP | -| vkWaitForFences | ⚙️ WIP | +| vkWaitForFences | ✅ Implemented | diff --git a/build.zig b/build.zig index b315a77..457d56b 100644 --- a/build.zig +++ b/build.zig @@ -595,7 +595,7 @@ fn addPhiDaemonCompilerArgs( switch (optimize) { .Debug => cmd.addArgs(&.{ "-O0", "-g3" }), .ReleaseSafe => cmd.addArgs(&.{ "-O2", "-g", "-DNDEBUG" }), - .ReleaseFast => cmd.addArgs(&.{ "-O3", "-DNDEBUG" }), + .ReleaseFast => cmd.addArgs(&.{ "-O3", "-DNDEBUG", "-DNOLOGS" }), .ReleaseSmall => cmd.addArgs(&.{ "-Os", "-DNDEBUG" }), } } @@ -612,6 +612,7 @@ fn addPhiDaemon(b: *std.Build, optimize: std.builtin.OptimizeMode, cc: []const u "src/phi/mic/Image.c", "src/phi/mic/Logger.c", "src/phi/mic/Memory.c", + "src/phi/mic/Queue.c", "src/phi/mic/Transport.c", // Add non-AVX files here }; diff --git a/src/phi/PhiBinarySemaphore.zig b/src/phi/PhiBinarySemaphore.zig index ce29589..21fab96 100644 --- a/src/phi/PhiBinarySemaphore.zig +++ b/src/phi/PhiBinarySemaphore.zig @@ -8,6 +8,10 @@ const Self = @This(); pub const Interface = base.BinarySemaphore; interface: Interface, +mutex: std.Io.Mutex, +condition: std.Io.Condition, +is_signaled: bool, +is_failed: bool, pub fn create(device: *base.Device, allocator: std.mem.Allocator, info: *const vk.SemaphoreCreateInfo) VkError!*Self { const self = allocator.create(Self) catch return VkError.OutOfHostMemory; @@ -23,6 +27,10 @@ pub fn create(device: *base.Device, allocator: std.mem.Allocator, info: *const v self.* = .{ .interface = interface, + .mutex = .init, + .condition = .init, + .is_signaled = false, + .is_failed = false, }; return self; } @@ -34,10 +42,39 @@ pub fn destroy(interface: *Interface, allocator: std.mem.Allocator) void { pub fn signal(interface: *Interface) VkError!void { const self: *Self = @alignCast(@fieldParentPtr("interface", interface)); - _ = self; + const io = interface.owner.io(); + + self.mutex.lock(io) catch return VkError.DeviceLost; + defer self.mutex.unlock(io); + + if (self.is_failed) return VkError.DeviceLost; + self.is_signaled = true; + self.condition.broadcast(io); +} + +/// Latch an asynchronous queue/device failure and wake all host waiters +pub fn fail(interface: *Interface) void { + const self: *Self = @alignCast(@fieldParentPtr("interface", interface)); + const io = interface.owner.io(); + + self.mutex.lock(io) catch return; + defer self.mutex.unlock(io); + + self.is_failed = true; + self.condition.broadcast(io); } pub fn wait(interface: *Interface) VkError!void { const self: *Self = @alignCast(@fieldParentPtr("interface", interface)); - _ = self; + const io = interface.owner.io(); + + self.mutex.lock(io) catch return VkError.DeviceLost; + defer self.mutex.unlock(io); + + while (!self.is_signaled and !self.is_failed) { + self.condition.wait(io, &self.mutex) catch return VkError.DeviceLost; + } + if (self.is_failed) return VkError.DeviceLost; + + self.is_signaled = false; } diff --git a/src/phi/PhiFence.zig b/src/phi/PhiFence.zig index 440a251..7bfa149 100644 --- a/src/phi/PhiFence.zig +++ b/src/phi/PhiFence.zig @@ -9,6 +9,10 @@ const Self = @This(); pub const Interface = base.Fence; interface: Interface, +mutex: std.Io.Mutex, +condition: std.Io.Condition, +is_signaled: bool, +is_failed: bool, pub fn create(device: *Device, allocator: std.mem.Allocator, info: *const vk.FenceCreateInfo) VkError!*Self { const self = allocator.create(Self) catch return VkError.OutOfHostMemory; @@ -26,6 +30,10 @@ pub fn create(device: *Device, allocator: std.mem.Allocator, info: *const vk.Fen self.* = .{ .interface = interface, + .mutex = .init, + .condition = .init, + .is_signaled = info.flags.signaled_bit, + .is_failed = false, }; return self; } @@ -37,21 +45,84 @@ pub fn destroy(interface: *Interface, allocator: std.mem.Allocator) void { pub fn getStatus(interface: *Interface) VkError!void { const self: *Self = @alignCast(@fieldParentPtr("interface", interface)); - _ = self; + const io = interface.owner.io(); + + self.mutex.lock(io) catch return VkError.DeviceLost; + defer self.mutex.unlock(io); + + if (self.is_failed) return VkError.DeviceLost; + if (!self.is_signaled) return VkError.NotReady; } pub fn reset(interface: *Interface) VkError!void { const self: *Self = @alignCast(@fieldParentPtr("interface", interface)); - _ = self; + const io = interface.owner.io(); + + self.mutex.lock(io) catch return VkError.DeviceLost; + defer self.mutex.unlock(io); + + if (self.is_failed) return VkError.DeviceLost; + self.is_signaled = false; } pub fn signal(interface: *Interface) VkError!void { const self: *Self = @alignCast(@fieldParentPtr("interface", interface)); - _ = self; + const io = interface.owner.io(); + + self.mutex.lock(io) catch return VkError.DeviceLost; + defer self.mutex.unlock(io); + + if (self.is_failed) return VkError.DeviceLost; + self.is_signaled = true; + self.condition.broadcast(io); +} + +/// Latch an asynchronous queue/device failure and wake all host waiters +pub fn fail(interface: *Interface) void { + const self: *Self = @alignCast(@fieldParentPtr("interface", interface)); + const io = interface.owner.io(); + + self.mutex.lock(io) catch return; + defer self.mutex.unlock(io); + + self.is_failed = true; + self.condition.broadcast(io); } pub fn wait(interface: *Interface, timeout: u64) VkError!void { const self: *Self = @alignCast(@fieldParentPtr("interface", interface)); - _ = self; - _ = timeout; + const io = interface.owner.io(); + + if (timeout == std.math.maxInt(@TypeOf(timeout))) { + self.mutex.lock(io) catch return VkError.DeviceLost; + defer self.mutex.unlock(io); + + while (!self.is_signaled and !self.is_failed) { + self.condition.wait(io, &self.mutex) catch return VkError.DeviceLost; + } + if (self.is_failed) return VkError.DeviceLost; + return; + } + + const deadline = std.Io.Clock.Timestamp.fromNow(io, .{ + .raw = .fromNanoseconds(@intCast(timeout)), + .clock = .awake, + }); + while (true) { + { + self.mutex.lock(io) catch return VkError.DeviceLost; + defer self.mutex.unlock(io); + + if (self.is_failed) return VkError.DeviceLost; + if (self.is_signaled) return; + } + + const remaining = deadline.durationFromNow(io); + if (remaining.raw.nanoseconds <= 0) return VkError.Timeout; + + (std.Io.Clock.Duration{ + .raw = .fromNanoseconds(@min(remaining.raw.nanoseconds, std.time.ns_per_ms)), + .clock = .awake, + }).sleep(io) catch return VkError.DeviceLost; + } } diff --git a/src/phi/PhiQueue.zig b/src/phi/PhiQueue.zig index 03ec1b2..7ff3b16 100644 --- a/src/phi/PhiQueue.zig +++ b/src/phi/PhiQueue.zig @@ -4,8 +4,10 @@ const base = @import("base"); const lib = @import("lib.zig"); const proto = lib.proto; +const PhiBinarySemaphore = @import("PhiBinarySemaphore.zig"); const PhiCommandBuffer = @import("PhiCommandBuffer.zig"); const PhiDevice = @import("PhiDevice.zig"); +const PhiFence = @import("PhiFence.zig"); const PhiTransport = @import("PhiTransport.zig"); const VkError = base.VkError; @@ -13,7 +15,55 @@ const VkError = base.VkError; const Self = @This(); pub const Interface = base.Queue; +const ring_capacity: usize = @intCast(proto.PHI_QUEUE_RING_CAPACITY); +const ring_capacity_u64: u64 = @intCast(ring_capacity); +const shutdown_sequence = std.math.maxInt(u64); + +const PreparedSubmit = struct { + wait_semaphores: std.ArrayList(*base.BinarySemaphore), + signal_semaphores: std.ArrayList(*base.BinarySemaphore), + command_backing: ?[]u8, + scif_offset: ?u64, + registered_size: usize, + command_size: usize, + command_count: u64, +}; + +const PendingCompletion = struct { + signal_semaphores: std.ArrayList(*base.BinarySemaphore), + fence: ?*base.Fence, + command_backing: ?[]u8, + scif_offset: ?u64, + registered_size: usize, +}; + +const TaskData = struct { + queue: *Self, + sequence: usize, + submits: std.ArrayList(PreparedSubmit), + fence: ?*base.Fence, +}; + interface: Interface, +transport: PhiTransport, +ring_backing: []u8, +ring_offset: u64, +shared: *proto.PhiQueueShared, + +submit_group: std.Io.Group, +completion_group: std.Io.Group, +mutex: std.Io.Mutex, +condition: std.Io.Condition, + +next_task_sequence: usize, +executing_task_sequence: usize, +next_remote_sequence: u64, +completed_sequence: u64, +pending: [ring_capacity]?PendingCompletion, + +error_state: ?VkError, +shutting_down: bool, +remote_stopped: bool, pub fn create(allocator: std.mem.Allocator, device: *base.Device, index: u32, family_index: u32, flags: vk.DeviceQueueCreateFlags) VkError!*Interface { const self = allocator.create(Self) catch return VkError.OutOfHostMemory; @@ -26,12 +76,118 @@ pub fn create(allocator: std.mem.Allocator, device: *base.Device, index: u32, fa .waitIdle = waitIdle, }; - self.* = .{ .interface = interface }; + const phi_device: *PhiDevice = @alignCast(@fieldParentPtr("interface", device)); + var transport = try phi_device.transport.connectPeer(); + errdefer transport.close(); + + const device_allocator = device.device_allocator.allocator(); + const page_size = std.heap.pageSize(); + const registered_size = std.mem.alignForward(usize, @sizeOf(proto.PhiQueueShared), page_size); + const ring_backing = device_allocator.alignedAlloc( + u8, + .fromByteUnits(std.heap.page_size_max), + registered_size, + ) catch return VkError.OutOfHostMemory; + errdefer device_allocator.free(ring_backing); + @memset(ring_backing, 0); + + const ring_offset = try transport.registerHostMemory(ring_backing); + errdefer transport.unregisterHostMemory(ring_offset, ring_backing.len) catch {}; + + const setup_request: proto.PhiQueueSetupRequest = .{ + .scif_offset = ring_offset, + .scif_size = ring_backing.len, + .ring_capacity = @intCast(ring_capacity), + .reserved = 0, + }; + var setup_reply = std.mem.zeroes(proto.PhiResultReply); + try transport.request( + proto.PHI_PACKET_QUEUE_SETUP, + std.mem.asBytes(&setup_request), + std.mem.asBytes(&setup_reply), + ); + if (setup_reply.result.status != proto.PHI_STATUS_OK) { + return PhiTransport.statusToErr(setup_reply.result.status); + } + + const shared: *proto.PhiQueueShared = @ptrCast(@alignCast(ring_backing.ptr)); + + self.* = .{ + .interface = interface, + .transport = transport, + .ring_backing = ring_backing, + .ring_offset = ring_offset, + .shared = shared, + .submit_group = .init, + .completion_group = .init, + .mutex = .init, + .condition = .init, + .next_task_sequence = 0, + .executing_task_sequence = 0, + .next_remote_sequence = 1, + .completed_sequence = 0, + .pending = [_]?PendingCompletion{null} ** ring_capacity, + .error_state = null, + .shutting_down = false, + .remote_stopped = false, + }; + + self.completion_group.async(device.io(), completionRunner, .{self}); return &self.interface; } pub fn destroy(interface: *Interface, allocator: std.mem.Allocator) VkError!void { const self: *Self = @alignCast(@fieldParentPtr("interface", interface)); + const io = interface.owner.io(); + const device_allocator = interface.owner.device_allocator.allocator(); + + waitIdle(interface) catch |err| { + std.log.scoped(.PhiQueue).warn("Queue did not become idle during destruction: {s}", .{@errorName(err)}); + }; + + var graceful_shutdown = true; + self.mutex.lock(io) catch { + graceful_shutdown = false; + }; + if (graceful_shutdown) { + self.shutting_down = true; + self.condition.broadcast(io); + self.mutex.unlock(io); + + self.transport.sendQueueDoorbell(shutdown_sequence) catch |err| { + graceful_shutdown = false; + std.log.scoped(.PhiQueue).warn("Failed to send queue shutdown doorbell: {s}", .{@errorName(err)}); + self.transport.close(); + }; + } else { + // Wake the blocking completion receiver before releasing queue storage + self.transport.close(); + } + + self.completion_group.await(io) catch |err| { + graceful_shutdown = false; + std.log.scoped(.PhiQueue).warn("Failed while joining completion receiver: {s}", .{@errorName(err)}); + }; + + if (graceful_shutdown) { + self.mutex.lock(io) catch { + graceful_shutdown = false; + }; + if (graceful_shutdown) { + graceful_shutdown = self.remote_stopped; + self.mutex.unlock(io); + } + } + + if (graceful_shutdown) { + self.transport.unregisterHostMemory(self.ring_offset, self.ring_backing.len) catch |err| { + std.log.scoped(.PhiQueue).warn("Failed to unregister queue ring: {s}", .{@errorName(err)}); + }; + } + + self.transport.close(); + cleanupPendingAfterClose(self, device_allocator); + device_allocator.free(self.ring_backing); allocator.destroy(self); } @@ -43,46 +199,470 @@ pub fn bindSparse(interface: *Interface, info: []const vk.BindSparseInfo, fence: } pub fn submit(interface: *Interface, infos: []Interface.SubmitInfo, fence: ?*base.Fence) VkError!void { - const device: *PhiDevice = @alignCast(@fieldParentPtr("interface", interface.owner)); + const self: *Self = @alignCast(@fieldParentPtr("interface", interface)); + const io = interface.owner.io(); + const allocator = interface.owner.device_allocator.allocator(); - for (infos) |info| { - for (info.wait_semaphores.items) |semaphore| { - try semaphore.wait(); - } + try self.checkHealthy(); - for (info.command_buffers.items) |command_buffer| { - const phi_command_buffer: *PhiCommandBuffer = @alignCast(@fieldParentPtr("interface", command_buffer)); + const data = allocator.create(TaskData) catch return VkError.OutOfDeviceMemory; + errdefer allocator.destroy(data); - const work_execution_request: proto.PhiWorkExecutionRequest = .{ - .cmd_count = phi_command_buffer.serialized_cmd_count, - .command_buffer_size = phi_command_buffer.commands.items.len, - }; - const payload_size = @sizeOf(proto.PhiWorkExecutionRequest) + phi_command_buffer.commands.items.len; - const allocator = interface.host_allocator.allocator(); - const payload = allocator.alloc(u8, payload_size) catch return VkError.OutOfHostMemory; - defer allocator.free(payload); + var prepared_submits = try prepareSubmits(self, allocator, infos); + errdefer deinitPreparedSubmits(self, allocator, &prepared_submits); - @memcpy(payload[0..@sizeOf(proto.PhiWorkExecutionRequest)], std.mem.asBytes(&work_execution_request)); - @memcpy(payload[@sizeOf(proto.PhiWorkExecutionRequest)..], phi_command_buffer.commands.items); + const sequence = blk: { + self.mutex.lock(io) catch return VkError.DeviceLost; + defer self.mutex.unlock(io); - // Synchronous queues for now - var reply = std.mem.zeroes(proto.PhiResultReply); - try device.transport.request(proto.PHI_PACKET_WORK_EXECUTION, payload, std.mem.asBytes(&reply)); + if (self.error_state) |err| return err; + if (self.shutting_down) return VkError.DeviceLost; - if (reply.result.status != proto.PHI_STATUS_OK) { - return PhiTransport.statusToErr(reply.result.status); - } - } + const value = self.next_task_sequence; + self.next_task_sequence += 1; + break :blk value; + }; - for (info.signal_semaphores.items) |semaphore| { - try semaphore.signal(); - } - } - if (fence) |value| { - try value.signal(); - } + data.* = .{ + .queue = self, + .sequence = sequence, + .submits = prepared_submits, + .fence = fence, + }; + + self.submit_group.async(io, taskRunner, .{data}); } pub fn waitIdle(interface: *Interface) VkError!void { - _ = interface; + const self: *Self = @alignCast(@fieldParentPtr("interface", interface)); + const io = interface.owner.io(); + + self.submit_group.await(io) catch { + self.markLost(VkError.DeviceLost); + return VkError.DeviceLost; + }; + + self.mutex.lock(io) catch return VkError.DeviceLost; + defer self.mutex.unlock(io); + + while (self.completed_sequence + 1 < self.next_remote_sequence and self.error_state == null) { + self.condition.wait(io, &self.mutex) catch return VkError.DeviceLost; + } + + if (self.error_state) |err| return err; +} + +fn checkHealthy(self: *Self) VkError!void { + const io = self.interface.owner.io(); + self.mutex.lock(io) catch return VkError.DeviceLost; + defer self.mutex.unlock(io); + + if (self.error_state) |err| return err; + if (self.shutting_down) return VkError.DeviceLost; +} + +fn prepareSubmits(self: *Self, allocator: std.mem.Allocator, infos: []Interface.SubmitInfo) VkError!std.ArrayList(PreparedSubmit) { + var submits = std.ArrayList(PreparedSubmit).initCapacity(allocator, infos.len) catch return VkError.OutOfDeviceMemory; + errdefer deinitPreparedSubmits(self, allocator, &submits); + + for (infos) |info| { + var prepared = try prepareSubmit(self, allocator, info); + submits.append(allocator, prepared) catch { + deinitPreparedSubmit(self, allocator, &prepared); + return VkError.OutOfDeviceMemory; + }; + } + + return submits; +} + +fn prepareSubmit(self: *Self, allocator: std.mem.Allocator, info: Interface.SubmitInfo) VkError!PreparedSubmit { + var wait_semaphores = info.wait_semaphores.clone(allocator) catch return VkError.OutOfDeviceMemory; + errdefer wait_semaphores.deinit(allocator); + + var signal_semaphores = info.signal_semaphores.clone(allocator) catch return VkError.OutOfDeviceMemory; + errdefer signal_semaphores.deinit(allocator); + + var command_size: usize = 0; + var command_count: u64 = 0; + for (info.command_buffers.items) |command_buffer| { + const phi_command_buffer: *PhiCommandBuffer = @alignCast(@fieldParentPtr("interface", command_buffer)); + + if (phi_command_buffer.commands.items.len > std.math.maxInt(usize) - command_size) { + return VkError.OutOfHostMemory; + } + command_size += phi_command_buffer.commands.items.len; + + const serialized_cmd_count: u64 = @intCast(phi_command_buffer.serialized_cmd_count); + if (serialized_cmd_count > std.math.maxInt(u64) - command_count) { + return VkError.OutOfHostMemory; + } + command_count += serialized_cmd_count; + } + + var command_backing: ?[]u8 = null; + var scif_offset: ?u64 = null; + var registered_size: usize = 0; + + if (command_size != 0) { + const page_size = std.heap.pageSize(); + if (command_size > std.math.maxInt(usize) - (page_size - 1)) { + return VkError.OutOfHostMemory; + } + registered_size = std.mem.alignForward(usize, command_size, page_size); + + const backing = allocator.alignedAlloc( + u8, + .fromByteUnits(std.heap.page_size_max), + registered_size, + ) catch return VkError.OutOfHostMemory; + errdefer allocator.free(backing); + @memset(backing, 0); + + var write_offset: usize = 0; + for (info.command_buffers.items) |command_buffer| { + const phi_command_buffer: *PhiCommandBuffer = @alignCast(@fieldParentPtr("interface", command_buffer)); + const commands = phi_command_buffer.commands.items; + @memcpy(backing[write_offset .. write_offset + commands.len], commands); + write_offset += commands.len; + } + + const offset = try self.transport.registerHostMemory(backing); + errdefer self.transport.unregisterHostMemory(offset, backing.len) catch {}; + + command_backing = backing; + scif_offset = offset; + } + + return .{ + .wait_semaphores = wait_semaphores, + .signal_semaphores = signal_semaphores, + .command_backing = command_backing, + .scif_offset = scif_offset, + .registered_size = registered_size, + .command_size = command_size, + .command_count = command_count, + }; +} + +fn deinitPreparedSubmits(self: *Self, allocator: std.mem.Allocator, submits: *std.ArrayList(PreparedSubmit)) void { + for (submits.items) |*prepared| { + deinitPreparedSubmit(self, allocator, prepared); + } + submits.deinit(allocator); +} + +fn deinitPreparedSubmit(self: *Self, allocator: std.mem.Allocator, prepared: *PreparedSubmit) void { + prepared.wait_semaphores.deinit(allocator); + prepared.signal_semaphores.deinit(allocator); + + if (prepared.scif_offset) |offset| { + self.transport.unregisterHostMemory(offset, prepared.registered_size) catch |err| { + std.log.scoped(.PhiQueue).warn("Failed to unregister staged command buffer: {s}", .{@errorName(err)}); + }; + } + if (prepared.command_backing) |backing| allocator.free(backing); + + prepared.command_backing = null; + prepared.scif_offset = null; + prepared.registered_size = 0; + prepared.command_size = 0; + prepared.command_count = 0; +} + +fn taskRunner(data: *TaskData) void { + const self = data.queue; + const io = self.interface.owner.io(); + const allocator = self.interface.owner.device_allocator.allocator(); + + defer { + deinitPreparedSubmits(self, allocator, &data.submits); + allocator.destroy(data); + } + + self.mutex.lock(io) catch { + failTask(data); + self.markLost(VkError.DeviceLost); + return; + }; + while (data.sequence != self.executing_task_sequence and self.error_state == null) { + self.condition.wait(io, &self.mutex) catch { + self.mutex.unlock(io); + failTask(data); + self.markLost(VkError.DeviceLost); + return; + }; + } + if (self.error_state != null) { + self.mutex.unlock(io); + failTask(data); + return; + } + self.mutex.unlock(io); + + var task_error: ?VkError = null; + + if (data.submits.items.len == 0) { + if (data.fence) |fence| { + var marker: PreparedSubmit = .{ + .wait_semaphores = .empty, + .signal_semaphores = .empty, + .command_backing = null, + .scif_offset = null, + .registered_size = 0, + .command_size = 0, + .command_count = 0, + }; + self.publish(&marker, fence) catch |err| { + task_error = err; + }; + if (task_error == null) data.fence = null; + } + } else { + for (data.submits.items, 0..) |*prepared, info_index| { + for (prepared.wait_semaphores.items) |semaphore| { + semaphore.wait() catch |err| { + task_error = err; + break; + }; + } + if (task_error != null) break; + + const submission_fence = if (info_index + 1 == data.submits.items.len) data.fence else null; + self.publish(prepared, submission_fence) catch |err| { + task_error = err; + break; + }; + if (submission_fence != null) data.fence = null; + } + } + + if (task_error) |err| { + failTask(data); + self.markLost(err); + return; + } + + self.mutex.lock(io) catch { + self.markLost(VkError.DeviceLost); + return; + }; + self.executing_task_sequence += 1; + self.condition.broadcast(io); + self.mutex.unlock(io); +} + +fn publish(self: *Self, prepared: *PreparedSubmit, fence: ?*base.Fence) VkError!void { + const io = self.interface.owner.io(); + + self.mutex.lock(io) catch return VkError.DeviceLost; + + while ((self.next_remote_sequence - 1) - self.completed_sequence >= ring_capacity_u64 and self.error_state == null) { + self.condition.wait(io, &self.mutex) catch { + self.mutex.unlock(io); + return VkError.DeviceLost; + }; + } + + if (self.error_state) |err| { + self.mutex.unlock(io); + return err; + } + if (self.shutting_down) { + self.mutex.unlock(io); + return VkError.DeviceLost; + } + + const sequence = self.next_remote_sequence; + if (sequence == shutdown_sequence) { + self.mutex.unlock(io); + return VkError.DeviceLost; + } + const slot: usize = @intCast((sequence - 1) % ring_capacity_u64); + if (self.pending[slot] != null) { + self.mutex.unlock(io); + return VkError.DeviceLost; + } + + const command_scif_offset = prepared.scif_offset orelse 0; + self.pending[slot] = .{ + .signal_semaphores = prepared.signal_semaphores, + .fence = fence, + .command_backing = prepared.command_backing, + .scif_offset = prepared.scif_offset, + .registered_size = prepared.registered_size, + }; + + prepared.signal_semaphores = .empty; + prepared.command_backing = null; + prepared.scif_offset = null; + prepared.registered_size = 0; + + self.shared.submissions[slot] = .{ + .sequence = sequence, + .command_scif_offset = command_scif_offset, + .command_size = prepared.command_size, + .command_count = prepared.command_count, + }; + + @atomicStore( + @TypeOf(self.shared.producer_sequence), + &self.shared.producer_sequence, + @intCast(sequence), + .release, + ); + self.next_remote_sequence += 1; + self.mutex.unlock(io); + + self.transport.sendQueueDoorbell(sequence) catch |err| { + self.markLost(VkError.DeviceLost); + return err; + }; +} + +fn completionRunner(self: *Self) void { + while (true) { + const completion = self.transport.receiveQueueCompletion() catch { + if (!self.isShuttingDown()) self.markLost(VkError.DeviceLost); + return; + }; + + if (completion.sequence == shutdown_sequence) { + const io = self.interface.owner.io(); + self.mutex.lock(io) catch return; + self.remote_stopped = completion.status == proto.PHI_STATUS_OK; + self.condition.broadcast(io); + self.mutex.unlock(io); + return; + } + + self.completeOne(completion); + } +} + +fn completeOne(self: *Self, completion: proto.PhiQueueCompletion) void { + const io = self.interface.owner.io(); + const allocator = self.interface.owner.device_allocator.allocator(); + + self.mutex.lock(io) catch { + self.markLost(VkError.DeviceLost); + return; + }; + + if (completion.sequence != self.completed_sequence + 1 or completion.sequence >= self.next_remote_sequence) { + self.mutex.unlock(io); + self.markLost(VkError.DeviceLost); + return; + } + + const slot: usize = @intCast((completion.sequence - 1) % ring_capacity_u64); + var pending = self.pending[slot] orelse { + self.mutex.unlock(io); + self.markLost(VkError.DeviceLost); + return; + }; + self.pending[slot] = null; + self.mutex.unlock(io); + + var cleanup_failed = false; + if (pending.scif_offset) |offset| { + self.transport.unregisterHostMemory(offset, pending.registered_size) catch |err| { + cleanup_failed = true; + std.log.scoped(.PhiQueue).err("Failed to unregister completed command buffer: {s}", .{@errorName(err)}); + }; + } + if (pending.command_backing) |backing| allocator.free(backing); + + if (completion.status != proto.PHI_STATUS_OK or cleanup_failed) { + self.markLost(VkError.DeviceLost); + failPending(&pending); + } else if (self.hasError()) { + failPending(&pending); + } else { + signalPending(self, &pending); + } + + pending.signal_semaphores.deinit(allocator); + + self.mutex.lock(io) catch { + self.markLost(VkError.DeviceLost); + return; + }; + self.completed_sequence = completion.sequence; + self.condition.broadcast(io); + self.mutex.unlock(io); +} + +fn signalPending(self: *Self, pending: *PendingCompletion) void { + var signal_failed = false; + + for (pending.signal_semaphores.items) |semaphore| { + semaphore.signal() catch { + signal_failed = true; + }; + } + if (pending.fence) |fence| { + fence.signal() catch { + signal_failed = true; + }; + } + + if (signal_failed) self.markLost(VkError.DeviceLost); +} + +fn failPending(pending: *PendingCompletion) void { + for (pending.signal_semaphores.items) |semaphore| { + PhiBinarySemaphore.fail(semaphore); + } + if (pending.fence) |fence| PhiFence.fail(fence); +} + +fn failTask(data: *TaskData) void { + for (data.submits.items) |*prepared| { + for (prepared.signal_semaphores.items) |semaphore| { + PhiBinarySemaphore.fail(semaphore); + } + } + if (data.fence) |fence| PhiFence.fail(fence); +} + +fn markLost(self: *Self, _: VkError) void { + const io = self.interface.owner.io(); + self.mutex.lock(io) catch return; + defer self.mutex.unlock(io); + + const first_failure = self.error_state == null; + self.error_state = VkError.DeviceLost; + self.condition.broadcast(io); + + if (!first_failure) return; + + for (&self.pending) |*entry| { + if (entry.*) |*pending| failPending(pending); + } +} + +fn hasError(self: *Self) bool { + const io = self.interface.owner.io(); + self.mutex.lock(io) catch return true; + defer self.mutex.unlock(io); + return self.error_state != null; +} + +fn isShuttingDown(self: *Self) bool { + const io = self.interface.owner.io(); + self.mutex.lock(io) catch return true; + defer self.mutex.unlock(io); + return self.shutting_down; +} + +fn cleanupPendingAfterClose(self: *Self, allocator: std.mem.Allocator) void { + for (&self.pending) |*entry| { + if (entry.*) |*pending| { + failPending(pending); + if (pending.command_backing) |backing| allocator.free(backing); + pending.signal_semaphores.deinit(allocator); + entry.* = null; + } + } } diff --git a/src/phi/PhiTransport.zig b/src/phi/PhiTransport.zig index 0e5bc6e..a550359 100644 --- a/src/phi/PhiTransport.zig +++ b/src/phi/PhiTransport.zig @@ -13,6 +13,7 @@ epd: Endpoint, sequence: u64 = 1, mutex: std.Io.Mutex = .init, instance: *base.Instance, +node_id: u16, pub fn init(instance: *base.Instance, node_id: u16) VkError!Self { const epd = blk: { @@ -45,6 +46,7 @@ pub fn init(instance: *base.Instance, node_id: u16) VkError!Self { var self: Self = .{ .epd = epd, .instance = instance, + .node_id = node_id, }; try self.handshake(); @@ -52,17 +54,30 @@ pub fn init(instance: *base.Instance, node_id: u16) VkError!Self { return self; } +pub fn connectPeer(self: *const Self) VkError!Self { + return init(self.instance, self.node_id); +} + pub fn deinit(self: *Self) void { var reply: proto.PhiResult = undefined; self.request(proto.PHI_PACKET_SHUTDOWN, &.{}, std.mem.asBytes(&reply)) catch |err| { std.log.scoped(.PhiTransport).warn("Failed to shut down remote session: {s}", .{@errorName(err)}); }; - closeEndpoint(self.epd); - scif.unload(); + self.close(); std.log.scoped(.PhiTransport).info("Closed connection", .{}); } +/// Close a transport without issuing an RPC shutdown. Queue transports switch +/// to a raw full-duplex doorbell protocol after setup and must use this path +pub fn close(self: *Self) void { + if (self.epd < 0) return; + + closeEndpoint(self.epd); + self.epd = -1; + scif.unload(); +} + pub fn request(self: *Self, command: c_uint, payload: []const u8, reply_payload: []u8) VkError!void { self.mutex.lock(self.instance.io()) catch return VkError.DeviceLost; defer self.mutex.unlock(self.instance.io()); @@ -98,6 +113,20 @@ pub fn request(self: *Self, command: c_uint, payload: []const u8, reply_payload: try self.readAll(reply_payload); } +pub fn sendQueueDoorbell(self: *Self, sequence: u64) VkError!void { + const doorbell: proto.PhiQueueDoorbell = .{ + .sequence = sequence, + }; + try self.writeAll(std.mem.asBytes(&doorbell)); +} + +pub fn receiveQueueCompletion(self: *Self) VkError!proto.PhiQueueCompletion { + // SAFETY: readAll initializes the complete structure. + var completion: proto.PhiQueueCompletion = undefined; + try self.readAll(std.mem.asBytes(&completion)); + return completion; +} + pub fn statusToErr(status: c_int) VkError { return switch (status) { proto.PHI_STATUS_OUT_OF_MEMORY => VkError.OutOfDeviceMemory, @@ -112,7 +141,7 @@ fn writeAll(self: *Self, bytes: []const u8) VkError!void { while (offset < bytes.len) { const written = scif.send(self.epd, bytes[offset..].ptr, bytes.len - offset, scif.send_block); if (written <= 0) { - return VkError.InitializationFailed; + return VkError.DeviceLost; } offset += @intCast(written); } @@ -123,7 +152,7 @@ fn readAll(self: *Self, bytes: []u8) VkError!void { while (offset < bytes.len) { const read = scif.recv(self.epd, bytes[offset..].ptr, bytes.len - offset, scif.recv_block); if (read <= 0) { - return VkError.InitializationFailed; + return VkError.DeviceLost; } offset += @intCast(read); } diff --git a/src/phi/mic/CommandBuffer.c b/src/phi/mic/CommandBuffer.c index bb88c6a..7c4f894 100644 --- a/src/phi/mic/CommandBuffer.c +++ b/src/phi/mic/CommandBuffer.c @@ -4,6 +4,8 @@ #include #include +#include + static const char* CommandName[] = { "CopyBuffer", "FillBuffer", "CopyBufferToImage", "CopyImageToBuffer", "CopyImage", }; @@ -13,8 +15,15 @@ PhiStatus ReadCommandData(PhiCommandReader* reader, void* data, uint64_t size) if(reader->remaining < size) return PHI_STATUS_BAD_MESSAGE; - if(ReadAll(reader->endpoint, data, (size_t)size) < 0) + if(reader->memory != NULL) + { + memcpy(data, reader->memory, (size_t)size); + reader->memory += (size_t)size; + } + else if(ReadAll(reader->endpoint, data, (size_t)size) < 0) + { return PHI_STATUS_BAD_MESSAGE; + } reader->remaining -= size; return PHI_STATUS_OK; @@ -25,6 +34,13 @@ int DrainCommandReader(PhiCommandReader* reader) if(reader->remaining == 0) return 0; + if(reader->memory != NULL) + { + reader->memory += (size_t)reader->remaining; + reader->remaining = 0; + return 0; + } + int result = DrainPayload(reader->endpoint, reader->remaining); reader->remaining = 0; return result; @@ -53,6 +69,45 @@ static PhiStatus ExecuteCommand(PhiCommandReader* reader, const PhiCmdHeader* co return PHI_STATUS_BAD_MESSAGE; } +static PhiStatus ExecuteCommands(PhiCommandReader* reader, uint64_t cmd_count) +{ + PhiStatus status = PHI_STATUS_OK; + + for(uint64_t cmd_index = 0; cmd_index < cmd_count; ++cmd_index) + { + PhiCmdHeader cmd_header; + status = ReadCommandHeader(reader, &cmd_header); + if(status != PHI_STATUS_OK) + break; + + status = ExecuteCommand(reader, &cmd_header); + if(status != PHI_STATUS_OK) + { + const size_t command_name_count = sizeof(CommandName) / sizeof(CommandName[0]); + const char* command_name = cmd_header.type < command_name_count ? CommandName[cmd_header.type] : "Unknown"; + LogErrorFmt("Command %s execution failed: %s", command_name, StatusName[status]); + break; + } + } + + return status; +} + +PhiStatus ExecuteCommandBuffer(const void* data, uint64_t size, uint64_t cmd_count) +{ + PhiCommandReader reader = { + .endpoint = PHI_ENDPOINT_INVALID, + .memory = data, + .remaining = size, + }; + + const PhiStatus status = ExecuteCommands(&reader, cmd_count); + if(status != PHI_STATUS_OK) + return status; + + return reader.remaining == 0 ? PHI_STATUS_OK : PHI_STATUS_BAD_MESSAGE; +} + int HandleWorkExecution(PhiEndpoint endpoint, const PhiMessageHeader* header) { PhiWorkExecutionRequest request; @@ -75,6 +130,7 @@ int HandleWorkExecution(PhiEndpoint endpoint, const PhiMessageHeader* header) PhiCommandReader reader = { .endpoint = endpoint, + .memory = NULL, .remaining = header->payload_size - sizeof(request), }; @@ -86,20 +142,7 @@ int HandleWorkExecution(PhiEndpoint endpoint, const PhiMessageHeader* header) return SendReply(endpoint, header, &reply, sizeof(reply)); } - for(uint64_t cmd_index = 0; cmd_index < request.cmd_count; ++cmd_index) - { - PhiCmdHeader cmd_header; - reply.result.status = ReadCommandHeader(&reader, &cmd_header); - if(reply.result.status != PHI_STATUS_OK) - break; - - reply.result.status = ExecuteCommand(&reader, &cmd_header); - if(reply.result.status != PHI_STATUS_OK) - { - LogErrorFmt("Command %s execution failed: %s", CommandName[cmd_header.type], StatusName[reply.result.status]); - break; - } - } + reply.result.status = ExecuteCommands(&reader, request.cmd_count); if(reader.remaining > 0 && DrainCommandReader(&reader) < 0) return -1; diff --git a/src/phi/mic/CommandBuffer.h b/src/phi/mic/CommandBuffer.h index ec5fdf4..cf63ba4 100644 --- a/src/phi/mic/CommandBuffer.h +++ b/src/phi/mic/CommandBuffer.h @@ -6,11 +6,13 @@ typedef struct PhiCommandReader { PhiEndpoint endpoint; + const uint8_t* memory; uint64_t remaining; } PhiCommandReader; int HandleWorkExecution(PhiEndpoint endpoint, const PhiMessageHeader* header); int DrainCommandReader(PhiCommandReader* reader); +PhiStatus ExecuteCommandBuffer(const void* data, uint64_t size, uint64_t cmd_count); PhiStatus ReadCommandData(PhiCommandReader* reader, void* data, uint64_t size); #endif diff --git a/src/phi/mic/Daemon.c b/src/phi/mic/Daemon.c index 42024ff..7108322 100644 --- a/src/phi/mic/Daemon.c +++ b/src/phi/mic/Daemon.c @@ -3,6 +3,7 @@ #include #include #include +#include static int HandleHello(PhiEndpoint endpoint, const PhiMessageHeader* header) { @@ -39,7 +40,10 @@ PhiEndpoint StartDaemon(void) PhiEndpoint endpoint = TransportListen(PHI_TRANSPORT_PORT); if(endpoint == PHI_ENDPOINT_INVALID) + { LogError("Could not listen on the Phi transport"); + return 0; + } LogInfo("Daemon started"); return endpoint; @@ -92,6 +96,9 @@ int HandlePacket(PhiEndpoint endpoint) return -1; break; + case PHI_PACKET_QUEUE_SETUP: + return HandleQueueSetup(endpoint, &header); + case PHI_PACKET_SHUTDOWN: if(DrainPayload(endpoint, header.payload_size) < 0) return -1; diff --git a/src/phi/mic/Logger.c b/src/phi/mic/Logger.c index 4ed8a67..29a6679 100644 --- a/src/phi/mic/Logger.c +++ b/src/phi/mic/Logger.c @@ -32,13 +32,15 @@ #define UNDERLINE_OFF 24 #define INVERSE_OFF 27 -inline static void SetConsoleColor(FILE* file, int code) +static inline void SetConsoleColor(FILE* file, int code) { fprintf(file, "\033[1;%dm", code); } void Log(LogLevel level, const char* fmt, const char* file, const char* function, int line, ...) { +#ifndef NOLOGS + time_t now = time(0); struct tm tstruct = *localtime(&now); char buffer[128]; @@ -86,11 +88,13 @@ void Log(LogLevel level, const char* fmt, const char* file, const char* function fputc('\n', out); fflush(out); +#endif + if(level == PHI_LOG_LEVEL_FATAL) { - SetConsoleColor(out, BG_RED); - fprintf(out, "Fatal Error: emergency exit\n"); - SetConsoleColor(out, BG_DEF); + SetConsoleColor(stderr, BG_RED); + fprintf(stderr, "Fatal Error: emergency exit\n"); + SetConsoleColor(stderr, BG_DEF); abort(); } } diff --git a/src/phi/mic/Queue.c b/src/phi/mic/Queue.c new file mode 100644 index 0000000..f6df104 --- /dev/null +++ b/src/phi/mic/Queue.c @@ -0,0 +1,152 @@ +#include + +#include +#include +#include + +#include +#include +#include +#include +#include + +static PhiStatus ExecuteQueueSubmission(PhiEndpoint endpoint, const PhiQueueSubmission* submission) +{ + if(submission->command_size == 0) + return submission->command_count == 0 ? PHI_STATUS_OK : PHI_STATUS_BAD_MESSAGE; + + if(submission->command_count == 0) + return PHI_STATUS_BAD_MESSAGE; + + void* commands = malloc((size_t)submission->command_size); + if(commands == NULL) + return PHI_STATUS_OUT_OF_MEMORY; + + PhiStatus status = PHI_STATUS_OK; + if(TransportReadRemote(endpoint, commands, (size_t)submission->command_size, submission->command_scif_offset) < 0) + { + LogErrorFmt("Failed to read from host: %s", strerror(errno)); + status = PHI_STATUS_INVALID_HANDLE; + } + else + status = ExecuteCommandBuffer(commands, submission->command_size, submission->command_count); + + free(commands); + return status; +} + +static int SendQueueCompletion(PhiEndpoint endpoint, uint64_t sequence, PhiStatus status) +{ + const PhiQueueCompletion completion = { + .sequence = sequence, + .status = status, + .reserved = 0, + }; + return WriteAll(endpoint, &completion, sizeof(completion)); +} + +// Returns 1 for a graceful queue shutdown, 0 when the peer disconnects, and -1 for a transport failure +static int RunQueue(PhiEndpoint endpoint, volatile PhiQueueShared* shared) +{ + uint64_t next_sequence = 1; + PhiStatus fatal_status = PHI_STATUS_OK; + + for(;;) + { + PhiQueueDoorbell doorbell; + if(ReadAll(endpoint, &doorbell, sizeof(doorbell)) < 0) + return 0; + + if(doorbell.sequence == PHI_QUEUE_SHUTDOWN_SEQUENCE) + return 1; + + if(doorbell.sequence < next_sequence) + continue; + + while(next_sequence <= doorbell.sequence) + { + const size_t slot = (size_t)((next_sequence - 1u) % PHI_QUEUE_RING_CAPACITY); + + __atomic_thread_fence(__ATOMIC_ACQUIRE); + const volatile PhiQueueSubmission* remote_submission = &shared->submissions[slot]; + const PhiQueueSubmission submission = { + .sequence = remote_submission->sequence, + .command_scif_offset = remote_submission->command_scif_offset, + .command_size = remote_submission->command_size, + .command_count = remote_submission->command_count, + }; + + PhiStatus status = fatal_status; + if(status == PHI_STATUS_OK) + { + if(submission.sequence != next_sequence) + status = PHI_STATUS_BAD_MESSAGE; + else + status = ExecuteQueueSubmission(endpoint, &submission); + } + + if(status != PHI_STATUS_OK && fatal_status == PHI_STATUS_OK) + { + LogErrorFmt("Queue submission %llu failed: %s", (unsigned long long)next_sequence, StatusName[status]); + fatal_status = status; + } + + __atomic_store_n(&shared->completed_sequence, next_sequence, __ATOMIC_RELEASE); + if(SendQueueCompletion(endpoint, next_sequence, status) < 0) + return -1; + + ++next_sequence; + } + } +} + +int HandleQueueSetup(PhiEndpoint endpoint, const PhiMessageHeader* header) +{ + PhiQueueSetupRequest request; + PhiResultReply reply = { + .result = { + .status = PHI_STATUS_OK, + .reserved = 0, + }, + }; + + if(header->payload_size != sizeof(request)) + { + if(DrainPayload(endpoint, header->payload_size) < 0) + return -1; + reply.result.status = PHI_STATUS_BAD_MESSAGE; + return SendReply(endpoint, header, &reply, sizeof(reply)); + } + + if(ReadAll(endpoint, &request, sizeof(request)) < 0) + return -1; + + if(request.ring_capacity != PHI_QUEUE_RING_CAPACITY || request.scif_size < sizeof(PhiQueueShared)) + { + reply.result.status = PHI_STATUS_INVALID_ARGUMENT; + return SendReply(endpoint, header, &reply, sizeof(reply)); + } + + volatile PhiQueueShared* shared = + scif_mmap(NULL, (size_t)request.scif_size, PROT_READ | PROT_WRITE, 0, endpoint, request.scif_offset); + if(shared == MAP_FAILED) + { + reply.result.status = PHI_STATUS_MAP_HOST_MEMORY_FAILED; + return SendReply(endpoint, header, &reply, sizeof(reply)); + } + + if(SendReply(endpoint, header, &reply, sizeof(reply)) < 0) + { + scif_munmap((void*)shared, (size_t)request.scif_size); + return -1; + } + + const int run_result = RunQueue(endpoint, shared); + if(scif_munmap((void*)shared, (size_t)request.scif_size) != 0) + return -1; + + if(run_result == 1) + return SendQueueCompletion(endpoint, PHI_QUEUE_SHUTDOWN_SEQUENCE, PHI_STATUS_OK); + + return run_result; +} diff --git a/src/phi/mic/Queue.h b/src/phi/mic/Queue.h new file mode 100644 index 0000000..d57375f --- /dev/null +++ b/src/phi/mic/Queue.h @@ -0,0 +1,9 @@ +#ifndef APE_PHI_QUEUE_H +#define APE_PHI_QUEUE_H + +#include +#include + +int HandleQueueSetup(PhiEndpoint endpoint, const PhiMessageHeader* header); + +#endif diff --git a/src/phi/mic/Transport.c b/src/phi/mic/Transport.c index 905d533..29d3703 100644 --- a/src/phi/mic/Transport.c +++ b/src/phi/mic/Transport.c @@ -28,6 +28,13 @@ PhiEndpoint TransportListen(uint16_t port) return endpoint; } +int TransportReadRemote(PhiEndpoint endpoint, void* data, size_t size, uint64_t remote_offset) +{ + if(size == 0) + return 0; + return scif_vreadfrom(endpoint, data, size, (off_t)remote_offset, SCIF_RMA_SYNC); +} + ssize_t TransportReceive(PhiEndpoint endpoint, void* data, size_t size) { return scif_recv(endpoint, data, size, SCIF_RECV_BLOCK); diff --git a/src/phi/mic/Transport.h b/src/phi/mic/Transport.h index 77e65f2..6411006 100644 --- a/src/phi/mic/Transport.h +++ b/src/phi/mic/Transport.h @@ -14,6 +14,7 @@ PhiEndpoint TransportAccept(PhiEndpoint endpoint); int TransportClose(PhiEndpoint endpoint); PhiEndpoint TransportListen(uint16_t port); +int TransportReadRemote(PhiEndpoint endpoint, void* data, size_t size, uint64_t remote_offset); ssize_t TransportReceive(PhiEndpoint endpoint, void* data, size_t size); ssize_t TransportSend(PhiEndpoint endpoint, const void* data, size_t size); diff --git a/src/phi/shared/Protocol.h b/src/phi/shared/Protocol.h index d61115f..3ea7bd7 100644 --- a/src/phi/shared/Protocol.h +++ b/src/phi/shared/Protocol.h @@ -7,9 +7,12 @@ #define PHI_MEMORY_ALIGNMENT 64 #define PHI_PROTOCOL_MAGIC 0x50484941u -#define PHI_PROTOCOL_VERSION 1u +#define PHI_PROTOCOL_VERSION 2u #define PHI_SCIF_PORT 43616u +#define PHI_QUEUE_RING_CAPACITY 64u +#define PHI_QUEUE_SHUTDOWN_SEQUENCE UINT64_MAX + #ifndef PHI_TRANSPORT_PORT #define PHI_TRANSPORT_PORT PHI_SCIF_PORT #endif @@ -24,6 +27,7 @@ typedef enum PhiPacketType PHI_PACKET_WORK_EXECUTION = 6, PHI_PACKET_SHUTDOWN = 7, PHI_PACKET_MAP_HOST_MEMORY = 8, + PHI_PACKET_QUEUE_SETUP = 9, } PhiPacketType; // When adding status, update StatusName in Logger.h @@ -104,4 +108,45 @@ typedef struct PhiWorkExecutionRequest uint64_t command_buffer_size; } PhiWorkExecutionRequest; +typedef struct PhiQueueSubmission +{ + uint64_t sequence; + uint64_t command_scif_offset; + uint64_t command_size; + uint64_t command_count; +} PhiQueueSubmission; + +typedef struct PhiQueueShared +{ + // Host-written producer timeline. Keep it on its own cache line + uint64_t producer_sequence; + uint8_t producer_padding[56]; + + // MIC-written completion timeline. Keep it on its own cache line + uint64_t completed_sequence; + uint8_t completion_padding[56]; + + PhiQueueSubmission submissions[PHI_QUEUE_RING_CAPACITY]; +} PhiQueueShared; + +typedef struct PhiQueueSetupRequest +{ + uint64_t scif_offset; + uint64_t scif_size; + uint32_t ring_capacity; + uint32_t reserved; +} PhiQueueSetupRequest; + +typedef struct PhiQueueDoorbell +{ + uint64_t sequence; +} PhiQueueDoorbell; + +typedef struct PhiQueueCompletion +{ + uint64_t sequence; + int32_t status; + uint32_t reserved; +} PhiQueueCompletion; + #endif diff --git a/src/vulkan/Queue.zig b/src/vulkan/Queue.zig index 44eb478..493679d 100644 --- a/src/vulkan/Queue.zig +++ b/src/vulkan/Queue.zig @@ -106,25 +106,11 @@ pub inline fn bindSparse(self: *Self, info: []const vk.BindSparseInfo, fence: ?* } pub fn submit(self: *Self, infos: []const vk.SubmitInfo, p_fence: ?*Fence) VkError!void { - if (infos.len == 0) { - if (p_fence) |fence| { - try fence.signal(); - } - return; - } - const allocator = self.host_allocator.cloneWithScope(.command).allocator(); var submit_infos = try SubmitInfo.initBlob(allocator, infos); defer SubmitInfo.deinitBlob(allocator, &submit_infos); - if (submit_infos.items.len == 0) { - if (p_fence) |fence| { - try fence.signal(); - } - return; - } - try self.dispatch_table.submit(self, submit_infos.items, p_fence); }