dcrubro/miniboinc
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Works

Browse Source
This commit is contained in:
DcruBro
2026-01-18 18:42:28 +01:00
parent 8ed280126c
commit ce70bf02bf
9 changed files with 260 additions and 143 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
include/common/packettype.h
View File

@@ -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

8
include/common/task/task.h
View File

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

1
include/common/tcpd/tcpclient.h
View File

@@ -18,6 +18,7 @@ struct TcpClient {
unsigned char dataBuf[MTU];
ssize_t dataBufLen;
long assignedTaskNumber;
void (*on_data)(struct TcpClient* client);
void (*on_disconnect)(struct TcpClient* client);

1
include/common/tcpd/tcpserver.h
View File

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

174
src/client/main.c
View File

@@ -12,6 +12,8 @@
#include <pthread.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <string.h>
#include <common/tcpd/tcpserver.h>
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);
// 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);
DynArr* execArgs = DYNARR_CREATE(char*, 4);
// 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);
printf("Executing task binary: %s\n", execName);
// 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;
//}
// 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;
}

6
src/common/task/task.c
View File

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

26
src/common/tcpd/tcpserver.c
View File

@@ -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;
}
}
}

178
src/server/main.c
View File

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

BIN
test_computation
View File

Binary file not shown.