node.c

/* Author : Thed Arthur TD, January 2026 */

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <stdbool.h>
#include <arpa/inet.h>
#include <sys/types.h> 
#include <sys/socket.h>
#include <pthread.h>
#include "node.h"
#include <signal.h>
#include <errno.h>

#define BUFFER_SIZE 1024

volatile sig_atomic_t keep_running = 1;

void handle_sigint(int sig){
    (void)sig; 
    printf("\n[Master] SIGINT received, shutting down...\n");
    keep_running = 0;
}

void *heartbeat_thread(void *arg){
    Heartbeat *hb = (Heartbeat *)arg;
    const char *msg = "HEARTBEAT";

    while(keep_running){
        sendto(hb->socket_pt, msg, strlen(msg), 0,
               (struct sockaddr *)&hb->registry_addr, sizeof(hb->registry_addr));
        sleep(2);
    }
    return NULL;
}

int main(int argc, char *argv[]){
    if(argc < 3){
        printf("Usage : %s <port> <registry_port>\n", argv[0]);
        exit(EXIT_FAILURE);
    }

    int PORT = atoi(argv[1]);
    int PORT_DEST = atoi(argv[2]);

    int socket_pt = socket(AF_INET, SOCK_DGRAM, 0);
    if(socket_pt < 0){ 
        perror("socket"); exit(EXIT_FAILURE); 
    }

    struct sockaddr_in servaddr, cliaddr, registry_addr;
    memset(&servaddr, 0, sizeof(servaddr));
    servaddr.sin_family = AF_INET;
    servaddr.sin_addr.s_addr = INADDR_ANY;
    servaddr.sin_port = htons(PORT);
    bind(socket_pt, (struct sockaddr *)&servaddr, sizeof(servaddr));

    memset(&registry_addr, 0, sizeof(registry_addr));
    registry_addr.sin_family = AF_INET;
    registry_addr.sin_port = htons(PORT_DEST);
    inet_pton(AF_INET, "127.0.0.1", &registry_addr.sin_addr);

    pthread_t hb_thread;
    Heartbeat hb_args = {socket_pt, registry_addr};
    pthread_create(&hb_thread, NULL, heartbeat_thread, &hb_args);

    const char *join_msg = "JOIN";
    sendto(socket_pt, join_msg, strlen(join_msg), 0,
           (struct sockaddr *)&registry_addr, sizeof(registry_addr));

    struct timeval tv;
    tv.tv_sec = 1;
    tv.tv_usec = 0;

    if(setsockopt(socket_pt, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv)) < 0){
        perror("setsockopt SO_RCVTIMEO");
    }

    printf("Worker listening on port %d...\n", PORT);

    signal(SIGINT, handle_sigint);

    char buffer[BUFFER_SIZE];
    while(keep_running){
        socklen_t len = sizeof(cliaddr);
        ssize_t n = recvfrom(socket_pt, buffer, BUFFER_SIZE - 1, 0,
                     (struct sockaddr *)&cliaddr, &len);

        if(n < 0){
            if(errno == EAGAIN || errno == EWOULDBLOCK){
                continue;
            }
            else{
                perror("recvfrom");
                break;
            }
        }

        buffer[n] = '\0';

        char ip[INET_ADDRSTRLEN];
        inet_ntop(AF_INET, &cliaddr.sin_addr, ip, sizeof(ip));

        if(strncmp(buffer, "TASK:", 5) == 0){
            char task_id[32], op[16];
            int a, b;
            if(sscanf(buffer, "TASK:%31[^:]:%15[^:]:%d:%d", task_id, op, &a, &b) == 4){
                int res = 0;
                char result_msg[64];

                if(strcmp(op,"ADD")==0){ 
                    res = a+b; 
                }
                else if(strcmp(op,"SUB")==0){ 
                    res = a-b; 
                }
                else if(strcmp(op,"MUL")==0){ 
                    printf("Performing heavy MUL %d * %d...\n", a, b);
                    sleep(3);
                    res = a*b;
                } 
                else{
                    snprintf(result_msg, sizeof(result_msg), "ERROR:UNKNOWN_OP");
                    sendto(socket_pt, result_msg, strlen(result_msg), 0,
                           (struct sockaddr *)&cliaddr, len);
                    continue;
                }

                snprintf(result_msg, sizeof(result_msg), "RESULT:%s:%d", task_id, res);
                sendto(socket_pt, result_msg, strlen(result_msg), 0,
                       (struct sockaddr *)&cliaddr, len);
                printf("Task %s %d %d = %d sent back to %s:%d\n",
                       op, a, b, res, ip, ntohs(cliaddr.sin_port));
            }
        } 
        else{
            printf("[%s:%d] %s\n", ip, ntohs(cliaddr.sin_port), buffer);
        }
    }

    pthread_join(hb_thread, NULL);
    close(socket_pt);
    return 0;
}