[Phi] implementing fully async queues, fences and semaphores

src/phi/mic/CommandBuffer.c

@@ -4,6 +4,8 @@
 #include <Buffer.h>
 #include <Image.h>
 
+#include <string.h>
+
 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;

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

src/phi/mic/Daemon.c

@@ -3,6 +3,7 @@
 #include <Daemon.h>
 #include <Logger.h>
 #include <Memory.h>
+#include <Queue.h>
 
 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;

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();
 	}
 }

src/phi/mic/Queue.c

@@ -0,0 +1,152 @@
+#include <Queue.h>
+
+#include <CommandBuffer.h>
+#include <Daemon.h>
+#include <Logger.h>
+
+#include <errno.h>
+#include <stdint.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/mman.h>
+
+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;
+}

src/phi/mic/Queue.h

@@ -0,0 +1,9 @@
+#ifndef APE_PHI_QUEUE_H
+#define APE_PHI_QUEUE_H
+
+#include <Protocol.h>
+#include <Transport.h>
+
+int HandleQueueSetup(PhiEndpoint endpoint, const PhiMessageHeader* header);
+
+#endif

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);

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);