diff --git a/include/common/packettype.h b/include/common/packettype.h index 54d3d76..58aeb81 100644 --- a/include/common/packettype.h +++ b/include/common/packettype.h @@ -11,7 +11,8 @@ typedef enum { PACKET_TYPE_TASK_REQUEST = 6, PACKET_TYPE_MISSING_INFO = 7, PACKET_TYPE_ACKNOWLEDGE = 8, - PACKET_TYPE_TASK_REJECT = 9 + PACKET_TYPE_TASK_REJECT = 9, + PACKET_TYPE_TASK_NONE_AVAILABLE = 10 } PacketType; #endif diff --git a/include/common/task/task.h b/include/common/task/task.h index 060455d..4e3c6d7 100644 --- a/include/common/task/task.h +++ b/include/common/task/task.h @@ -26,10 +26,10 @@ typedef enum { TASK_ERR_INVTASK } task_err; -typedef struct { - char name[32]; // string - DynArr* data; -} task_arg_t; +//typedef struct { +// char name[32]; // string +// DynArr* data; +//} task_arg_t; typedef struct { uint32_t taskId; diff --git a/include/common/tcpd/tcpclient.h b/include/common/tcpd/tcpclient.h index 882081c..af560bf 100644 --- a/include/common/tcpd/tcpclient.h +++ b/include/common/tcpd/tcpclient.h @@ -17,7 +17,8 @@ struct TcpClient { uint32_t clientId; unsigned char dataBuf[MTU]; - ssize_t dataBufLen; + ssize_t dataBufLen; + long assignedTaskNumber; void (*on_data)(struct TcpClient* client); void (*on_disconnect)(struct TcpClient* client); diff --git a/include/common/tcpd/tcpserver.h b/include/common/tcpd/tcpserver.h index 3216e16..5da346a 100644 --- a/include/common/tcpd/tcpserver.h +++ b/include/common/tcpd/tcpserver.h @@ -48,6 +48,7 @@ void TcpServer_Init(TcpServer* ptr, unsigned short port, const char* addr); void TcpServer_Start(TcpServer* ptr, int maxcons); void TcpServer_Stop(TcpServer* ptr); void TcpServer_Send(TcpServer* ptr, TcpClient* cli, void* data, size_t len); +void Generic_SendSocket(int sock, void* data, size_t len); void TcpServer_Disconnect(TcpServer* ptr, TcpClient* cli); void TcpServer_KillClient(TcpServer* ptr, TcpClient* cli); diff --git a/src/client/main.c b/src/client/main.c index a771d13..5ebd835 100644 --- a/src/client/main.c +++ b/src/client/main.c @@ -12,6 +12,8 @@ #include #include #include +#include +#include char dataBuf[1500]; DynArr* capabilities; @@ -71,12 +73,42 @@ void on_data(ssize_t dataBufLen) { } break; } - case PACKET_TYPE_TASK_ASSIGN: - task_t task; - memcpy(&task, &dataBuf[1], sizeof(task_t)); - printf("Received TASK_ASSIGN packet from server for task %u\n", task.taskId); + case PACKET_TYPE_TASK_ASSIGN: { + uint32_t taskId; + char binary[64]; + uint32_t packetSize; + uint8_t* rawArgs = DYNARR_CREATE(uint8_t, 1); - DynArr* execArgs = DYNARR_CREATE(char*, 4); + // Parse task + uint8_t* ptr = (uint8_t*)&dataBuf[1]; // Skip packet type byte + memcpy(&packetSize, ptr, sizeof(uint32_t)); + ptr += sizeof(uint32_t); + memcpy(&taskId, ptr, sizeof(uint32_t)); + ptr += sizeof(uint32_t); + memcpy(&binary, ptr, 64); + ptr += 64; + + // Copy the data into the array + uint32_t argsSize = packetSize - (sizeof(uint32_t) + 64); + uint8_t argsData[argsSize]; + memcpy(argsData, ptr, argsSize); + + // Dump the args to a file (input.bin) + FILE* file = fopen("input.bin", "wb"); + if (!file) { + perror("Failed to open input.bin file!"); + exit(1); // Crash + } + + fwrite(argsData, argsSize, 1, file); + fclose(file); + + //task_t task; + //memcpy(&task, &dataBuf[1], sizeof(task_t)); + printf("Received TASK_ASSIGN packet from server for task %u\n", taskId); + printf(" Binary: %s\n", binary); + printf(" Data Size: %u\n", packetSize); + // RESULT report is ACKNOWLEDGE for now // Fork with binary name pid_t pid = fork(); @@ -86,84 +118,16 @@ void on_data(ssize_t dataBufLen) { return; } - printf("Data Size: %zu\n", dataBufLen); - - char** argv = NULL; - if (pid == 0) { - // Child with args - // Jump to special sequence to find args "ARGSTART" - size_t offset = 1 + sizeof(task_t); - while (offset < dataBufLen - 8) { - if (memcmp(&dataBuf[offset], "ARGSTART", 8) == 0) { - offset += 8; - break; - } - offset++; - } + char execName[strlen(binary) + 3]; + snprintf(execName, sizeof(execName), "./%s", binary); // TODO; Improve - printf("Args Offset: %zu\n", offset); - printf("Args length: %zu\n", dataBufLen - offset); - - // Parse args - DynArr_push_back(execArgs, task.binary); // First arg is binary name - while (offset < dataBufLen) { - //if (memcmp(&dataBuf[offset], "ARGNEXT", 7) == 0) { - // offset += 7; - // continue; - //} + printf("Executing task binary: %s\n", execName); - // Read raw arg as 32 bytes (name) - //char argName[32]; - //memcpy(argName, &dataBuf[offset], 32); - //offset += 32; - //printf("Arg Name: %s\n", argName); - - // Ignore name for now; just pass raw data - offset += 32; - - // Find position of next ARGNEXT or end - size_t argDataStart = offset; - size_t argDataEnd = dataBufLen;; - for (size_t i = offset; i < dataBufLen - 7; i++) { - if (memcmp(&dataBuf[i], "ARGNEXT", 7) == 0) { - argDataEnd = i; - break; - } - } - - if (argDataEnd <= argDataStart) { - printf("No more arg data found\n"); - break; - } - - printf("Arg Data Start: %zu\n", argDataStart); - printf("Arg Data End: %zu\n", argDataEnd); - - size_t argDataLen = argDataEnd - argDataStart; - printf("Arg Data Length: %zu\n", argDataLen); - - char* argData = malloc(argDataLen); - memcpy(argData, &dataBuf[argDataStart], argDataLen); // No null-termination; this is raw data - - offset += argDataLen; - } - DynArr_push_back(execArgs, NULL); // Null-terminate args array - - // Convert into char** - argv = malloc(DynArr_size(execArgs) * sizeof(char*) + 1); - for (size_t i = 0; i < DynArr_size(execArgs); i++) { - char** argPtr = (char**)DynArr_at(execArgs, i); - argv[i] = *argPtr; - } - argv[DynArr_size(execArgs)] = NULL; // Null-terminate - - printf("Executing task %u: %s\n", task.taskId, task.binary); - char execName[66]; - snprintf(execName, sizeof(execName), "./%s", task.binary); // TODO; Improve + char* argv[] = { execName, NULL }; // Exec - execvp(execName, (char* const*)argv); + execvp(execName, argv); // If execvp returns, there was an error perror("execvp failed"); exit(1); // Crash child @@ -175,7 +139,7 @@ void on_data(ssize_t dataBufLen) { if (WIFEXITED(status)) { int exitStatus = WEXITSTATUS(status); - printf("Task %u completed with exit status %d\n", task.taskId, exitStatus); + printf("Task %u completed with exit status %d\n", taskId, exitStatus); if (exitStatus == 0) { // Send TASK_RESULT packet @@ -188,6 +152,16 @@ void on_data(ssize_t dataBufLen) { if (!outFile) { perror("Failed to open output.bin"); } + + uint32_t size = 0; // Size of the output.bin excl. EOF + fseek(outFile, 0, SEEK_END); + size = ftell(outFile); + fseek(outFile, 0, SEEK_SET); + // Push size + for (size_t i = 0; i < sizeof(uint32_t); i++) { + DynArr_push_back(packet, &((uint8_t*)&size)[i]); + } + int byte; while ((byte = fgetc(outFile)) != EOF) { uint8_t b = (uint8_t)byte; @@ -196,29 +170,46 @@ void on_data(ssize_t dataBufLen) { fclose(outFile); // Send - send(sock, packet->data, DynArr_size(packet) * sizeof(uint8_t), 0); - printf("Sent TASK_RESULT packet for task %u to server\n", task.taskId); + Generic_SendSocket(sock, packet->data, DynArr_size(packet) * sizeof(uint8_t)); + LAST_PACKET_SENT = PACKET_TYPE_TASK_RESULT; + printf("Sent TASK_RESULT packet for task %u to server\n", taskId); DynArr_destroy(packet); + + // Get another task after some time + sleep(2); // Wait 2 seconds before requesting again + char taskRequestPacket[1]; + taskRequestPacket[0] = PACKET_TYPE_TASK_REQUEST; + Generic_SendSocket(sock, taskRequestPacket, 1); + printf("Requested task from server\n"); + LAST_PACKET_SENT = PACKET_TYPE_TASK_REQUEST; } else { - // TODO; Send TASK_REJECT packet to server to inform that task failed + // Send TASK_REJECT packet to server to inform that task failed + char rejectPacket[1]; + rejectPacket[0] = PACKET_TYPE_TASK_REJECT; + send(sock, rejectPacket, 1, 0); + printf("Sent TASK_REJECT packet for task %u to server\n", taskId); break; } } else { - printf("Task %u did not complete successfully\n", task.taskId); + printf("Task %u did not complete successfully\n", taskId); } } - for (size_t i = 0; i < DynArr_size(execArgs); i++) { - char** argPtr = (char**)DynArr_at(execArgs, i); - if (argPtr && *argPtr) { - free(*argPtr); - } - } + break; + } + case PACKET_TYPE_TASK_NONE_AVAILABLE: + printf("Server indicates no tasks are currently available\n"); + // Optionally wait and request again later + sleep(5); // Wait 5 seconds before requesting again - DynArr_destroy(execArgs); - free(argv); + // Request task again + char taskRequestPacket[1]; + taskRequestPacket[0] = PACKET_TYPE_TASK_REQUEST; + send(sock, taskRequestPacket, 1, 0); + printf("Requested task from server again\n"); + LAST_PACKET_SENT = PACKET_TYPE_TASK_REQUEST; break; default: @@ -290,4 +281,5 @@ int main(void) { close(sock); return 0; -} + +} \ No newline at end of file diff --git a/src/common/task/task.c b/src/common/task/task.c index 2db1d7d..d4297e5 100644 --- a/src/common/task/task.c +++ b/src/common/task/task.c @@ -15,7 +15,7 @@ void Task_Create(task_t* tsk) { tsk->assigned_to = 0; // ClientID 0 is reserved tsk->state = TASK_NONE; memset(tsk->binary, 0, 64); - tsk->args = DYNARR_CREATE(task_arg_t, 1); + tsk->args = DYNARR_CREATE(uint8_t, 1); tsk->results = DYNARR_CREATE(uint8_t, 1); } @@ -35,12 +35,12 @@ void Task_DestroyArgs(task_t* task) { return; } - for (size_t i = 0; i < DynArr_size(task->args); i++) { + /*for (size_t i = 0; i < DynArr_size(task->args); i++) { task_arg_t* arg = (task_arg_t*)DynArr_at(task->args, i); if (arg && arg->data) { DynArr_destroy(arg->data); } - } + }*/ DynArr_destroy(task->args); DynArr_destroy(task->results); diff --git a/src/common/tcpd/tcpserver.c b/src/common/tcpd/tcpserver.c index 6e7a16b..b175c92 100644 --- a/src/common/tcpd/tcpserver.c +++ b/src/common/tcpd/tcpserver.c @@ -156,6 +156,7 @@ void* TcpServer_threadprocess(void* ptr) { heapCli->clientId = random_four_byte(); heapCli->dataBufLen = 0; heapCli->capabilities = DYNARR_CREATE(capability_t, 4); + heapCli->assignedTaskNumber = -1; // None assigned size_t i; for (i = 0; i < svr->clients; i++) { @@ -247,7 +248,30 @@ void TcpServer_Stop(TcpServer* ptr) { void TcpServer_Send(TcpServer* ptr, TcpClient* cli, void* data, size_t len) { if (ptr && cli && data && len > 0) { - send(cli->clientFd, data, len, 0); + size_t sent = 0; + while (sent < len) { + // Ensure that all data is sent. TCP can split sends. + ssize_t n = send(cli->clientFd, (unsigned char*)data + sent, len - sent, 0); + if (n < 0) { + perror("tcpserver - send error"); + break; + } + sent += n; + } + } +} + +void Generic_SendSocket(int sock, void* data, size_t len) { + if (sock > 0 && data && len > 0) { + size_t sent = 0; + while (sent < len) { + ssize_t n = send(sock, (unsigned char*)data + sent, len - sent, 0); + if (n < 0) { + perror("generic - send socket error"); + break; + } + sent += n; + } } } @@ -298,4 +322,4 @@ size_t Generic_FindClientInArrayByPtr(TcpClient** arr, TcpClient* ptr, size_t le } return SIZE_MAX; // Returns max unsigned, likely improbable to be correct -} +} \ No newline at end of file diff --git a/src/server/main.c b/src/server/main.c index e08c944..12588d2 100644 --- a/src/server/main.c +++ b/src/server/main.c @@ -11,6 +11,7 @@ volatile int running = 1; task_queue_t taskQueue; +TcpServer* svr; void signalHandler(int sig) { printf("Caught signal %d, shutting down...\n", sig); @@ -22,7 +23,8 @@ void on_connect(TcpClient* client) { helloPacket[0] = PACKET_TYPE_HELLO; helloPacket[1] = SERVER_INFO_VERSION; memcpy(&helloPacket[2], SERVER_INFO_IDENTIFIER, 16); - send(client->clientFd, helloPacket, SERVER_HELLO_PACKET_SIZE + 1, 0); + TcpServer_Send(svr, client, helloPacket, SERVER_HELLO_PACKET_SIZE + 1); + //send(client->clientFd, helloPacket, SERVER_HELLO_PACKET_SIZE + 1, 0); printf("Client connected: %u\n", client->clientId); } @@ -62,7 +64,8 @@ void on_data(TcpClient* client) { // Send ACKNOWLEDGE; Might embed which packet is being acknowledged later char ackPacket[1]; ackPacket[0] = PACKET_TYPE_ACKNOWLEDGE; - send(client->clientFd, ackPacket, 1, 0); + TcpServer_Send(svr, client, ackPacket, 1); + //send(client->clientFd, ackPacket, 1, 0); printf("Sent ACKNOWLEDGE to client %u\n", client->clientId); } @@ -70,6 +73,12 @@ void on_data(TcpClient* client) { case PACKET_TYPE_TASK_REQUEST: printf("Received TASK_REQUEST packet from client %u\n", client->clientId); + if (client->assignedTaskNumber != -1) { + // Client already has a task assigned + printf("Client %u already has a task assigned (%ld), ignoring request\n", client->clientId, client->assignedTaskNumber); + break; + } + if (client->capabilities) { size_t capCount = DynArr_size(client->capabilities); @@ -78,7 +87,7 @@ void on_data(TcpClient* client) { char missingInfoPacket[2]; missingInfoPacket[0] = PACKET_TYPE_MISSING_INFO; missingInfoPacket[1] = PACKET_TYPE_CLIENT_CAPABILITIES; - send(client->clientFd, missingInfoPacket, 2, 0); + TcpServer_Send(svr, client, missingInfoPacket, 2); printf("Requested capabilities from client %u\n", client->clientId); break; } @@ -113,39 +122,33 @@ void on_data(TcpClient* client) { uint8_t packetType = PACKET_TYPE_TASK_ASSIGN; DynArr_push_back(packet, &packetType); - // Push back the task struct - for (size_t k = 0; k < sizeof(task_t); k++) { - DynArr_push_back(packet, ((char*)task) + k); + uint32_t dataSize = sizeof(uint32_t) + 64 + DynArr_size(task->args); + // Push data size + for (size_t k = 0; k < sizeof(uint32_t); k++) { + DynArr_push_back(packet, &((uint8_t*)&dataSize)[k]); } - // Special sequence to indicate start of args - char sequence[] = "ARGSTART"; - for (size_t k = 0; k < sizeof(sequence); k++) { - DynArr_push_back(packet, &sequence[k]); + // Task ID + for (size_t k = 0; k < sizeof(uint32_t); k++) { + DynArr_push_back(packet, &((uint8_t*)&task->taskId)[k]); } - // Push back args - for (size_t k = 0; k < DynArr_size(task->args); k++) { - task_arg_t* arg = (task_arg_t*)DynArr_at(task->args, k); - for (size_t m = 0; m < sizeof(task_arg_t); m++) { - DynArr_push_back(packet, ((char*)arg) + m); - } - // Push back arg data - for (size_t m = 0; m < DynArr_size(arg->data); m++) { - char* dataPtr = (char*)DynArr_at(arg->data, m); - for (size_t n = 0; n < sizeof(char); n++) { - DynArr_push_back(packet, dataPtr + n); - } - } - - // Push back next arg indicator - char nextArgIndicator[] = "ARGNEXT"; - for (size_t n = 0; n < sizeof(nextArgIndicator); n++) { - DynArr_push_back(packet, &nextArgIndicator[n]); - } + // Binary name + for (size_t k = 0; k < 64; k++) { + DynArr_push_back(packet, &task->binary[k]); // char is uint8_t compatible } - send(client->clientFd, packet->data, DynArr_size(packet) * sizeof(uint8_t), 0); + // Args + uint32_t argsSize = (uint32_t)DynArr_size(task->args); + for (size_t k = 0; k < argsSize; k++) { + uint8_t* argByte = (uint8_t*)DynArr_at(task->args, k); + DynArr_push_back(packet, argByte); + } + + TcpServer_Send(svr, client, packet->data, DynArr_size(packet) * sizeof(uint8_t)); + + client->assignedTaskNumber = task->taskId; + DynArr_destroy(packet); printf("Assigned task %u to client %u\n", task->taskId, client->clientId); break; @@ -157,12 +160,87 @@ void on_data(TcpClient* client) { } } } + + if (!assigned) { + // No task available + char noneAvailablePacket[1]; + noneAvailablePacket[0] = PACKET_TYPE_TASK_NONE_AVAILABLE; + TcpServer_Send(svr, client, noneAvailablePacket, 1); + printf("No available tasks for client %u\n", client->clientId); + } + break; } + break; + case PACKET_TYPE_TASK_REJECT: + printf("Received TASK_REJECT packet from client %u\n", client->clientId); + + if (client->assignedTaskNumber == -1) { + // No task assigned + printf("Client %u has no task assigned, ignoring reject\n", client->clientId); + break; + } else { + // Mark task as pending again + for (size_t i = 0; i < DynArr_size(taskQueue.tasks); i++) { + task_t* task = (task_t*)DynArr_at(taskQueue.tasks, i); + if (task && task->taskId == client->assignedTaskNumber) { + task->state = TASK_PENDING; + task->assigned_to = 0; + task->assigned_at = 0; + printf("Marked task %u as pending again\n", task->taskId); + break; + } + } + client->assignedTaskNumber = -1; // Clear assigned task + } + break; case PACKET_TYPE_TASK_RESULT: printf("Received TASK_RESULT packet from client %u\n", client->clientId); + + if (client->assignedTaskNumber == -1) { + // No task assigned + printf("Client %u has no task assigned, ignoring result\n", client->clientId); + break; + } else { + // Mark task as done; TODO: Store results + for (size_t i = 0; i < DynArr_size(taskQueue.tasks); i++) { + task_t* task = (task_t*)DynArr_at(taskQueue.tasks, i); + if (task && task->taskId == client->assignedTaskNumber) { + task->state = TASK_DONE; + printf("Marked task %u as done\n", task->taskId); + + // Temporary: log as int + //if (bytesReceived - 1 >= sizeof(int)) { + // int result; + // memcpy(&result, &client->dataBuf[1 + sizeof(uint32_t)], sizeof(int)); + // printf("Task %u result: %d\n", task->taskId, result); + //} + uint32_t resultDataSize; + memcpy(&resultDataSize, &client->dataBuf[1], sizeof(uint32_t)); + printf("Task %u result data size: %u bytes\n", task->taskId, resultDataSize); + + // Store results + task->assigned_to = 0; + task->assigned_at = 0; + task->results = DYNARR_CREATE(uint8_t, 1); + // Store result data + for (size_t j = 1; j < resultDataSize; j++) { + uint8_t byte = client->dataBuf[j + 1 + sizeof(uint32_t)]; // +1 for packet type, +4 for size + DynArr_push_back(task->results, &byte); + } + + // TODO: Optionally dump results to file or process them here + // Note: This assumes that the results are trustworthy, which might need to be verified via multiple clients + + break; + } + } + + client->assignedTaskNumber = -1; // Clear assigned task + } + break; default: printf("Received unknown packet type %u from client %u\n", packetType, client->clientId); @@ -183,21 +261,41 @@ int main(void) { TaskQueue_Init(&taskQueue); printf("Task Queue Created!\n"); - task_t task1; - Task_Create(&task1); - task1.state = TASK_PENDING; - task1.args = DYNARR_CREATE(task_arg_t, 1); - strcpy(task1.binary, "test_computation"); - task_arg_t* blk = DynArr_push_back(task1.args, &(task_arg_t){ .name = "input_data", .data = DYNARR_CREATE(char, 1) }); - DynArr_push_back(blk->data, "12"); // Compute the square of 12 - task1.taskId = 1; - TaskQueue_AddTask(&taskQueue, &task1); + //task_t task1; + //Task_Create(&task1); + //task1.state = TASK_PENDING; + //task1.args = DYNARR_CREATE(uint8_t, 1); + //int argVal = 10000019; // This should stay in the stack due to the loop below (I guess we'll see soon enough) + //strcpy(task1.binary, "test_computation"); + //for (size_t i = 0; i < sizeof(argVal); i++) { + // uint8_t b = ((uint8_t*)&argVal)[i]; // Byte hacking to get each byte of the int + // DynArr_push_back(task1.args, &b); + //} + //task1.taskId = 1; +// + //TaskQueue_AddTask(&taskQueue, &task1); + + // Create some dummy tasks + for (int i = 0; i < 5; i++) { + task_t task; + Task_Create(&task); + task.state = TASK_PENDING; + task.args = DYNARR_CREATE(uint8_t, 1); + int argVal = rand(); // Different arg for each task + snprintf(task.binary, sizeof(task.binary), "test_computation"); + for (size_t j = 0; j < sizeof(argVal); j++) { + uint8_t b = ((uint8_t*)&argVal)[j]; // Byte hacking to get each byte of the int + DynArr_push_back(task.args, &b); + } + TaskQueue_AddTask(&taskQueue, &task); // AddTask copies to the dynamic array, so we can free the stack task safely + } + //Task_DestroyArgs(&task1); printf("Task added!\n"); - TcpServer* svr = TcpServer_Create(); + svr = TcpServer_Create(); TcpServer_Init(svr, 6175, "127.0.0.1"); svr->on_connect = on_connect; svr->on_data = on_data; diff --git a/test_computation b/test_computation index 2b10b15..6e0ddc2 100755 Binary files a/test_computation and b/test_computation differ