Priority Job/Task Queue for Linux (sockets and multithreading)

Question

Preface

Please review my implementation of a job queue for linux/unix systems. This is my first time coding in C, although I have quite some experience in C++.

I know this is a moderate amount of loc, so please focus on jobq.c, jobq_server. c, manager.c, and server.c. I have uploaded the other files for context. I am mainly intersted in feedback in my design choices and possible memory management problems. Anything beyond this is also greatly appreciated, but not expected.

The setting of CPU affinities is ultimately done over taskset, although I feel I could have implemented that myself aswell, but it is what it is. Maybe in a future update.

Introduction

In the scenario where multiple people want to run calculations on the same machine, scheduling can become an issue. For multiprocessing applications, a single user can potentially block the entire computing capacity. The jobq server and client should do the following things:

• Allow each user to submit a job, specifying number of threads and a timeout after which their job will be forcefully terminated

• Allow configuration for maximum number of cores, max timelimits and additionally definition of a "long job", which may not acquire more than n cores.

• If the next job in the queue can not start, because too few cores are available, it will be marked as a priority element and start as soon as the current latest job has finished. (It could technically start sooner, but I was too lazy to implement a smarter solution. This solution always works, albeit it's not the best).

More specifically, this is what the /etc/jobq/.config file looks like

port=7331
host=127.0.0.1
maxcores=20
maxtime=432000
longjob=43200
longmaxcores=11

colours.h

#ifndef JOBQ_COLOURS_H
#define JOBQ_COLOURS_H

#define RED "\x1B[31m"
#define GREEN "\x1B[32m"
#define YELLOW "\x1B[33m"
#define BLUE "\x1B[34m"
#define MAGENTA "\x1B[35m"
#define CYAN "\x1B[36m"
#define WHITE "\x1B[37m"
#define RESET "\x1B[0m"


#endif //JOBQ_COLOURS_H

config.h

This just contains some constants and defines struct Config.

#ifndef JOBQ_CONFIG_H
#define JOBQ_CONFIG_H
#define CONFIG_FILE "/etc/jobq/.config"
#define DIRECTORY_BUFFER 256
#define USERNAME_BUFFER 32
#define ANSWER_BUFFER 4096
#define MAX_CMD_LENGTH 512
#define MAX_PENDING_CONNECTIONS 4
#define MSG_SUBMIT 'x'
#define MSG_STOP 'y'
#define MSG_STATUS 'z'

struct Config{
    long port;
    char server_ip[40];
    long maxcores;
    long maxtime;
    long longjob;
    long longmaxcores;
};
#endif //JOBQ_CONFIG_H

job.h

Defines struct Job and a doubly linked list.

#ifndef JOBQ_JOB_H
#define JOBQ_JOB_H

#include "config.h"
#include <time.h>
#include <unistd.h>

struct Job{
    long id;
    pid_t pid;
    long cores;
    long time_limit;
    time_t start_time;
    time_t end_time;
    uid_t user_id;
    gid_t group_id;
    char user_name[USERNAME_BUFFER];
    char working_directory[DIRECTORY_BUFFER];
    char cmd[MAX_CMD_LENGTH];
    unsigned long long int core_mask;
};

struct Job_List{
    struct Node{
        struct Job job;
        struct Node* next;
        struct Node* prev;
    } *first, *last;
};

struct Node* push_back(struct Job_List* this, struct Job j);
struct Node* erase(struct Job_List* this, struct Node** elem);


#endif //JOBQ_JOB_H

manager.h

This is a struct to manage communication between the two threads, so that I do not need to point to a million variables but just a single struct.

#ifndef JOBQ_MANAGER_H
#define JOBQ_MANAGER_H

#include <pthread.h>
#include <time.h>

struct Job_List;
struct Node;

struct Manager{
    struct Job_List* running_queue;
    struct Job_List* waiting_queue;
    struct Node* priority_job;
    time_t latest_end_time;
    unsigned long long int available_cores;
    pthread_mutex_t *running_lock;
    pthread_mutex_t *waiting_lock;
};

struct Node* start_job(struct Manager* m, struct Node* node);
void clear_finished_and_overdue_jobs(struct Manager* m);
time_t get_latest_end_time(struct Manager* m);
long get_free_cores(struct Manager* m);
void start_jobs(struct Manager* m);
#endif //JOBQ_MANAGER_H

parse_config.h

Just declarations

#ifndef JOBQ_PARSE_CONFIG_H
#define JOBQ_PARSE_CONFIG_H

struct Config;
void parse_config(const char* file_name, struct Config* config);

#endif //JOBQ_PARSE_CONFIG_H

queue.h

Just declarations

#ifndef JOBQ_QUEUE_H
#define JOBQ_QUEUE_H

void* queue(void* buffer);

#endif //JOBQ_QUEUE_H

server.h

Just declarations

#ifndef JOBQ_SERVER_H
#define JOBQ_SERVER_H

void* server(void* pointers);

#endif //JOBQ_SERVER_H

utils.h

Just some declarations

#ifndef JOBQ_UTIL_H
#define JOBQ_UTIL_H

int parse_long(const char* str, long* value);

void quit(const char* msg);
void quit_with_error(const char* msg);

char** split(char* str, char tok, int* len);
#endif //JOBQ_UTIL_H

jobq.c

Implementation of push and erase

#include "job.h"

#include <malloc.h>

struct Node* push_back(struct Job_List* this, struct Job j){
    struct Node *node = malloc(sizeof(struct Node));
    node->job = j;
    node->next = NULL;
    node->prev = NULL;
    if( this->first == NULL ){
        this->first = node;
        this->last = node;
    }else{
        node->prev = this->last;
        this->last->next = node;
        this->last = node;
    }
    return node;
}

struct Node* erase(struct Job_List* this, struct Node** node){
    if( this == NULL || node == NULL || *node == NULL ){
        return NULL;
    }
    if( (*node)->next != NULL ){
        (*node)->next->prev = (*node)->prev;
    }
    if( (*node)->prev != NULL ){
        (*node)->prev->next = (*node)->next;
    }
    if( this->first == *node ){
        this->first = (*node)->next;
    }
    if( this->last == *node ){
        this->last = (*node)->prev;
    }
    struct Node* next = (*node)->next;
    free(*node);
    *node = NULL;
    node = NULL;
    return next;
}

jobq.c (please review this)

This is the client for the application.

#include "config.h"
#include "job.h"
#include "parse_config.h"
#include "util.h"

#include <arpa/inet.h>
#include <stdio.h>
#include <string.h>
#include <sys/socket.h>
#include <unistd.h>
#include <stdlib.h>
#include <pwd.h>


#define CMD_SUBMIT "submit"
#define CMD_STOP "stop"
#define CMD_STATUS "status"



void print_usage(){
    printf("jobq usage:\n");
    printf("jobq %s <n_cores> <time_limit in seconds> \"<command>\"\n", CMD_SUBMIT);
    printf("jobq %s <job_id>\n", CMD_STOP);
    printf("jobq %s\n", CMD_STATUS);
    printf("Please note that the quotation marks are mandatory, if you want to supply arguments to your executable.\n");
}

void prepare_submit_message(char* message_buffer, const char** argv, struct Config* config){
    struct Job job;
    job.id = 0;
    job.pid = 0;
    job.user_id = geteuid();
    job.group_id = getegid();
    job.start_time = 0;
    job.end_time = 0;
    job.core_mask = 0;
    memset(&job.user_name, 0, USERNAME_BUFFER);
    memset(&job.working_directory, 0, DIRECTORY_BUFFER);
    memset(&job.cmd[0], 0, MAX_CMD_LENGTH);
    struct passwd *pass = getpwuid(getuid());
    if( pass == NULL ){
        quit_with_error("getpwuid failure");
    }
    memcpy(&job.user_name[0], (pass->pw_name), strlen(pass->pw_name));

    if( getcwd(&job.working_directory[0], DIRECTORY_BUFFER) == NULL ){
        printf("Working directory path must not exceed %i characters.", DIRECTORY_BUFFER);
        exit(EXIT_FAILURE);
    }

    if( !parse_long(argv[2], &job.cores) ){
        printf("Number of cores must be valid integer: %s\n", argv[2]);
        exit(EXIT_FAILURE);
    }
    if( job.cores < 1 || job.cores > config->maxcores ){
        printf("Number of cores must be 0 < n < %ld. Given: %ld\n", config->maxcores, job.cores);
        exit(EXIT_FAILURE);
    }
    if( !parse_long(argv[3], &job.time_limit) ){
        printf("Time limit must be valid integer: %s\n", argv[3]);
        exit(EXIT_FAILURE);
    }
    if( job.time_limit < 1 || job.time_limit > config->maxtime ){
        printf("Time limit must be 0 < n < %ld. Given: %ld\n", config->maxtime, job.time_limit);
        exit(EXIT_FAILURE);
    }
    if( job.time_limit > config->longjob && job.cores > config->longmaxcores ){
        printf("Jobs taking longer than %ld must not acquire more than %ld cores.\n", config->longjob, config->longmaxcores);
        exit(EXIT_FAILURE);
    }
    size_t cmd_length = strlen(argv[4]);
    if( cmd_length < 1 || cmd_length > MAX_CMD_LENGTH){
        printf("Command length must be between 1 < n < %d. Given: %lu", MAX_CMD_LENGTH, cmd_length);
        exit(EXIT_FAILURE);
    }
    memcpy(&job.cmd, argv[4], cmd_length);
    message_buffer[0] = MSG_SUBMIT;
    memcpy(&message_buffer[1], &job, sizeof(struct Job));
}

void prepare_status_message(char* message_buffer){
    message_buffer[0] = MSG_STATUS;
}

void prepare_stop_message(char* message_buffer, const char** argv){
    struct Job job;
    job.id = 0;
    job.pid = 0;
    job.user_id = geteuid();
    job.group_id = getegid();
    job.start_time = 0;
    job.end_time = 0;
    job.core_mask = 0;
    job.cores = 0;
    job.time_limit = 0;
    memset(&job.user_name, 0, USERNAME_BUFFER);
    memset(&job.working_directory, 0, DIRECTORY_BUFFER);
    memset(&job.cmd[0], 0, MAX_CMD_LENGTH);
    struct passwd *pass = getpwuid(getuid());
    if( pass == NULL ){
        quit_with_error("getpwuid failure");
    }
    memcpy(&job.user_name[0], (pass->pw_name), strlen(pass->pw_name));
    if( !parse_long(argv[2], &job.id) ){
        printf("jobq_id must be valid integer. Given: %s\n", argv[2]);
        exit(EXIT_FAILURE);
    }
    message_buffer[0] = MSG_STOP;
    memcpy(&message_buffer[1], &job, sizeof(struct Job));
}

enum Action{
    SUBMIT = 0,
    STOP = 1,
    STATUS = 2
};

int main(int argc, char const* argv[]){
    if( argc <= 1 ){
        print_usage();
        exit(EXIT_SUCCESS);
    }

    enum Action action;

    if( strcmp(argv[1], CMD_SUBMIT) == 0 && argc == 5 ){
        action = SUBMIT;
    }
    else if( strcmp(argv[1], CMD_STOP) == 0 && argc == 3 ){
        action = STOP;
    }
    else if( strcmp(argv[1], CMD_STATUS) == 0 && argc == 2){
        action = STATUS;
    }
    else{
        printf("Error: invalid arguments.\n");
        print_usage();
        exit(EXIT_FAILURE);
    }

    struct Config config;
    parse_config(CONFIG_FILE, &config);

    int socket_descriptor = socket(AF_INET, SOCK_STREAM, 0);
    if( socket_descriptor == -1 ){
        quit_with_error("Socket failure");
    }

    struct sockaddr_in server_address;
    server_address.sin_family = AF_INET;
    server_address.sin_port = htons(config.port);

    int inet = inet_pton(AF_INET, config.server_ip, &server_address.sin_addr);

    if( inet == 0 ){
        printf("%s is not a valid server ip.\n", config.server_ip);
        exit(EXIT_FAILURE);
    }
    else if( inet == -1 ){
        quit_with_error("Inet_pton failure");
    }

    int client_descriptor = connect(socket_descriptor, (struct sockaddr*)&server_address, sizeof(server_address));
    if( client_descriptor == -1 ){
        quit_with_error("Connect failure");
    }

    char message_buffer[sizeof(struct Job)] = {0};

    switch( action ){
        case SUBMIT:{
            prepare_submit_message(&message_buffer[0], argv, &config);
            break;
        }
        case STATUS:{
            prepare_status_message(&message_buffer[0]);
            break;
        }
        case STOP:{
            prepare_stop_message(&message_buffer[0], argv);
            break;
        }
        default:{
        }
    }

    ssize_t result = send(socket_descriptor, message_buffer, sizeof(struct Job), 0);
    if( result == -1 ){
        quit_with_error("Send failure");
    }

    char answer_buffer[ANSWER_BUFFER] = {0};
    size_t answer = read(socket_descriptor, answer_buffer, ANSWER_BUFFER);

    if( answer == -1 ){
        quit_with_error("Read failure");
    }

    puts(answer_buffer);
    close(client_descriptor);
    close(socket_descriptor);
    return EXIT_SUCCESS;
}

jobq_server (please review this)

The server for the application. This will start two threads as discussed above.

#include "config.h"
#include "job.h"
#include "manager.h"
#include "parse_config.h"
#include "queue.h"
#include "server.h"
#include "util.h"

#include <pthread.h>
#include <string.h>

#define MAX_PENDING_CONNECTIONS 4

int main(int argc, char** argv){
    char* message_buffer[sizeof(struct Job)+1];
    struct Config config;
    parse_config(CONFIG_FILE, &config);
    struct Job_List waiting_queue;
    waiting_queue.first = NULL;
    waiting_queue.last = NULL;
    struct Job_List running_queue;
    running_queue.first = NULL;
    running_queue.last = NULL;

    pthread_mutex_t running_lock;
    pthread_mutex_t waiting_lock;

    if( pthread_mutex_init(&running_lock, NULL) != 0){
        quit_with_error("Could not initialize mutex");
    }
    if( pthread_mutex_init(&waiting_lock, NULL) != 0){
        quit_with_error("Could not initialize mutex");
    }

    struct Manager manager;
    manager.running_queue = &running_queue;
    manager.waiting_queue = &waiting_queue;
    manager.priority_job = NULL;
    manager.latest_end_time = 0;
    manager.available_cores = 0xffffffffffffffffULL >> (64 - config.maxcores);
    manager.running_lock = &running_lock;
    manager.waiting_lock = &waiting_lock;


    void* pointers[4] = {(void*) message_buffer, (void*) &config, (void*) &manager};


    pthread_t socket_thread_id, queue_thread_id;
    pthread_create(&socket_thread_id, NULL, server, &pointers[0]);
    pthread_create(&queue_thread_id, NULL, queue, &pointers[0]);

    pthread_join(socket_thread_id, NULL);
    pthread_join(queue_thread_id, NULL);

    return EXIT_SUCCESS;
}

manager.c (please review this)

This manages cleans and updates the running queue of finished and overdue jobs, sets the priority element and loads new jobs from the waiting queue.

#include <fcntl.h>
#include <stdio.h>
#include <string.h>
#include <pthread.h>
#include <signal.h>
#include <sys/wait.h>
#include <stdlib.h>

#include "job.h"
#include "manager.h"
#include "util.h"

struct Node* start_job(struct Manager* m, struct Node* node){
    if( m == NULL || node == NULL || m->running_queue == NULL || m->waiting_queue == NULL ){
        return NULL;
    }
    node->job.core_mask = 0;
    unsigned long long int it = m->available_cores;
    for( int i = 0, j = 0; i < node->job.cores; j++, it >>= 1 ){
        if( (it & 1) == 1 ){
            node->job.core_mask |= (1ULL << j);
            i++;
        }
    }
    m->available_cores ^= node->job.core_mask;

    node->job.start_time = time(NULL);
    node->job.end_time = node->job.start_time + node->job.time_limit;

    node->job.pid = fork();
    if( node->job.pid == -1 ){
        fprintf(stderr, "Failed to fork\n");
    }
    else if( node->job.pid == 0 ){
        if( chdir(node->job.working_directory) < 0 ){
            quit_with_error("chdir failure");
        }
        if( setegid(node->job.group_id) < 0 ){
            quit_with_error("setegid failure");
        }
        if( setgid(node->job.group_id) < 0 ){
            quit_with_error("setgid failure");
        }
        if( setuid(node->job.user_id) < 0){
            quit_with_error("setuid failure");
        }
        if( seteuid(node->job.user_id) < 0){
            quit_with_error("seteuid failure");
        }
        char out[24] = {0};
        char err[24] = {0};
        sprintf(&out[0], "out_%ld.txt", node->job.id);
        sprintf(&err[0], "err_%ld.txt", node->job.id);
        int fd_out = open(out, O_WRONLY | O_CREAT, S_IRWXU);
        if( fd_out < 0 ){
            quit_with_error("error open stdout");
        }
        int fd_err = open(err, O_WRONLY | O_CREAT, S_IRWXU);
        if( fd_err < 0 ){
            quit_with_error("error open stderr");
        }
        if( dup2(fd_out, STDOUT_FILENO) < 0 ){
            quit_with_error("dup2 failure out");
        }
        if( dup2(fd_err, STDERR_FILENO) < 0 ){
            quit_with_error("dup2 failure err");
        }
        char* command = &node->job.cmd[0];
        int len = 0;
        char** split_str = split(command, ' ', &len);
        char* arr[] = {"taskset", "-a", NULL};
        int memory_len = snprintf(NULL,0,"0x%llX", node->job.core_mask);
        arr[2] = malloc(memory_len + 1);
        snprintf(arr[2], memory_len+1,"0x%llX", node->job.core_mask);
        char* params[3+len];
        memcpy(&params[0], &arr[0], 3 * sizeof(char*));
        memcpy(&params[3], &split_str[0], len * sizeof(char*));
        execv("/usr/bin/taskset", params);
    }
    char time_buffer[15] = {0};
    time_t now = time(NULL);
    strftime(&time_buffer[0], 15, "%d-%m %H:%M:%S", localtime(&now));
    fprintf(stderr, "[%s] Starting job: %ld %ld\n", &time_buffer[0], node->job.id, (long)node->job.pid);
    pthread_mutex_lock(m->running_lock);
    push_back(m->running_queue, node->job);
    pthread_mutex_unlock(m->running_lock);
    struct Node* next = erase(m->waiting_queue, &node);
    return next;
}

void clear_finished_and_overdue_jobs(struct Manager* m){
    if( m == NULL ){
        return;
    }
    pthread_mutex_lock(m->running_lock);
    struct Node* node = m->running_queue->first;
    while( node != NULL ){
        if( node->job.end_time < time(NULL) ){
            if( kill(node->job.pid, 9) < 0 ){
                quit_with_error("Could not kill process");
            }else{
                usleep(1000000);
                char time_buffer[15] = {0};
                time_t now = time(NULL);
                strftime(&time_buffer[0], 15, "%d-%m %H:%M:%S", localtime(&now));
                fprintf(stderr, "[%s] Terminated job: %ld %ld\n", &time_buffer[0], node->job.id, (long) node->job.pid);
            }
        }
        int status = 0;
        pid_t pid = waitpid(node->job.pid, &status, WNOHANG);
        if( pid == node->job.pid ){
            char time_buffer[15] = {0};
            time_t now = time(NULL);
            strftime(&time_buffer[0], 15, "%d-%m %H:%M:%S", localtime(&now));
            fprintf(stderr,"[%s] Job finished: %ld %ld.\n", &time_buffer[0], node->job.id, (long)node->job.pid);
            m->available_cores |= node->job.core_mask;
            node = erase(m->running_queue, &node);
        }else{
            node = node->next;
        }
    }
    pthread_mutex_unlock(m->running_lock);
}


time_t get_latest_end_time(struct Manager* m){
    time_t latest = 0;
    pthread_mutex_lock(m->running_lock);
    struct Node* node = m->running_queue->first;
    while( node != NULL ){
        if( node->job.end_time > latest ){
            latest = node->job.end_time;
        }
        node = node->next;
    }
    pthread_mutex_unlock(m->running_lock);
    return latest;
}

long get_free_cores(struct Manager* m){
    if( m == NULL ){
        return 0;
    }
    long result = 0;
    unsigned long long int n = m->available_cores;
    while( n ){
        result += (long)(n&1ULL);
        n>>=1;
    }
    return result;
}

void start_jobs(struct Manager* m){
    long n_free_cores = get_free_cores(m);
    if( m->priority_job != NULL && m->priority_job->job.cores <= n_free_cores ){
        start_job(m, m->priority_job);
        m->latest_end_time = get_latest_end_time(m);
        m->priority_job = NULL;
    }
    pthread_mutex_lock(m->waiting_lock);
    struct Node *waiting_node = m->waiting_queue->first;
    while( waiting_node != NULL)
    {
        if( m->priority_job == NULL )
        {
            if( waiting_node->job.cores <= get_free_cores(m) )
            {
                waiting_node = start_job(m, waiting_node);
                m->latest_end_time = get_latest_end_time(m);
            }else
            {
                m->priority_job = malloc(sizeof(struct Node));
                if( m->priority_job == NULL ){
                    fprintf(stderr, "Could not allocate for priority node\n");
                    exit(EXIT_FAILURE);
                }
                *m->priority_job = *waiting_node;
                m->priority_job->job.start_time = m->latest_end_time;
                m->priority_job->job.end_time = m->latest_end_time + m->priority_job->job.time_limit;
                waiting_node = erase(m->waiting_queue, &waiting_node);
            }
        }else{
            if( waiting_node->job.time_limit + time(NULL) < m->priority_job->job.start_time && waiting_node->job.cores <= get_free_cores(m) )
            {
                waiting_node = start_job(m, waiting_node);
            }else{
                waiting_node = waiting_node->next;
            }
        }
    }
    pthread_mutex_unlock(m->waiting_lock);
}

queue.c

Thread that calls the functions from manager.c

#include "job.h"
#include "manager.h"
#include "queue.h"

#include <unistd.h>
#include <stdbool.h>
#include <string.h>


void *queue(void *pointers){
    void** p = (void**) pointers;
    struct Manager* m = p[2];

    while( true ){
        clear_finished_and_overdue_jobs(m);
        start_jobs(m);
        sleep(1);
    }
    return NULL;
}

server.c (please review this)

This checks for incoming requests of jobq and sends out status messages, puts jobs into the waiting queue or stops them / removes them.

#include "colours.h"
#include "config.h"
#include "job.h"
#include "manager.h"
#include "server.h"
#include "time.h"
#include "util.h"

#include <netinet/in.h> // struct socket_in
#include <sys/socket.h>

#include <malloc.h>
#include <pthread.h>
#include <stdbool.h>
#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include <signal.h>

void* server(void* pointers){
    void** p = (void**) pointers;
    struct Config* config = p[1];
    struct Manager* m = p[2];
    int socket_descriptor = socket(AF_INET, SOCK_STREAM, 0);
    if( socket_descriptor == -1 ){
        quit_with_error("Socket failure");
    }

    struct sockaddr_in socket_address;
    int option = 1;
    if( setsockopt(socket_descriptor, SOL_SOCKET, SO_REUSEADDR | SO_REUSEPORT, &option, sizeof(option))) {
        quit_with_error("Setsockopt failure");
    }
    socket_address.sin_family = AF_INET;
    socket_address.sin_addr.s_addr = INADDR_ANY;
    socket_address.sin_port = htons(config->port);

    if( bind(socket_descriptor, (struct sockaddr*)&socket_address, sizeof(socket_address)) == -1 ){
        quit_with_error("Bind failure");
    }

    if( listen(socket_descriptor, MAX_PENDING_CONNECTIONS) == -1 ){
        quit_with_error("Listen failure");
    }


    socklen_t socket_address_length = sizeof(socket_address);


    long job_id = 0;
    while( true ){
        int temp_descriptor = accept(socket_descriptor, (struct sockaddr*)&socket_address, &socket_address_length);

        char buffer[sizeof(struct Job)+1] = {0};
        ssize_t read_bytes = read(temp_descriptor, &buffer, sizeof(struct Job));

        if(read_bytes < 0 ){
            quit_with_error("Read failure.");
        }
        if( read_bytes == 0 ){
            continue;
        }
        if( buffer[0] == MSG_SUBMIT ){
            struct Job job;
            memcpy(&job, &buffer[1], sizeof(struct Job));
            job.id = job_id++;
            pthread_mutex_lock(m->waiting_lock);
            char time_buffer[15] = {0};
            time_t now = time(NULL);
            strftime(&time_buffer[0], 15, "%d-%m %H:%M:%S", localtime(&now));
            fprintf(stderr, "[%s] Adding job to waiting_queue: %ld\n", &time_buffer[0], job.id);
            push_back(m->waiting_queue, job);
            pthread_mutex_unlock(m->waiting_lock);
            char answer_buffer[ANSWER_BUFFER] = {0};
            sprintf(&answer_buffer[0], "Job submitted. Id: %ld\nEnter jobq status to check status.\n", job.id);
            send(temp_descriptor, answer_buffer, strlen(answer_buffer), 0);
        }
        if( buffer[0] == MSG_STOP ){
            char answer_buffer[ANSWER_BUFFER] = {0};
            struct Job job;
            memcpy(&job, &buffer[1], sizeof(struct Job));
            int found_job = 0;
            {
                pthread_mutex_lock(m->running_lock);
                struct Node *r_node = m->running_queue->first;
                while( r_node != NULL)
                {
                    if( r_node->job.id == job.id )
                    {
                        found_job = 1;
                        if( r_node->job.user_id != job.user_id )
                        {
                            sprintf(&answer_buffer[0], RED "Error: Can not stop job %ld. Insufficient permissions. (Not your job)\n", job.id);
                        }else
                        {
                            if( kill(r_node->job.pid, 9) < 0 )
                            {
                                quit_with_error("Could not kill process.");
                            }else
                            {
                                usleep(1000000);
                                char time_buffer[15] = {0};
                                time_t now = time(NULL);
                                strftime(&time_buffer[0], 15, "%d-%m %H:%M:%S", localtime(&now));
                                fprintf(stderr, "[%s] User stopped job: %ld %ld\n", &time_buffer[0], r_node->job.id, (long) r_node->job.pid);
                                m->available_cores |= r_node->job.core_mask;
                                erase(m->running_queue, &r_node);
                                sprintf(&answer_buffer[0], GREEN "Success" RESET ": Stopped job %ld\n", job.id);
                            }
                        }
                        break;
                    }
                    r_node = r_node->next;
                }
                pthread_mutex_unlock(m->running_lock);
            }
            pthread_mutex_lock(m->waiting_lock);
            if( !found_job )
            {
                if( m->priority_job != NULL)
                {
                    if( m->priority_job->job.id == job.id )
                    {
                        found_job = 1;
                        if( m->priority_job->job.user_id != job.user_id )
                        {
                            sprintf(&answer_buffer[0], RED "Error" RESET ": Can not stop job %ld. Insufficient permissions. (Not your job)\n", job.id);
                        }else
                        {
                            char time_buffer[15] = {0};
                            time_t now = time(NULL);
                            strftime(&time_buffer[0], 15, "%d-%m %H:%M:%S", localtime(&now));
                            fprintf(stderr, "[%s] User stopped job: %ld %ld\n", &time_buffer[0], m->priority_job->job.id, (long) m->priority_job->job.pid);
                            m->priority_job = NULL;
                            sprintf(&answer_buffer[0], GREEN "Success" RESET ": Stopped job %ld\n", job.id);
                        }
                    }
                }
            }
            if( !found_job ){
                struct Node *w_node = m->waiting_queue->first;
                while( w_node != NULL)
                {
                    if( w_node->job.id == job.id )
                    {
                        found_job = 1;
                        if( w_node->job.user_id != job.user_id )
                        {
                            sprintf(&answer_buffer[0], RED "Error: Can not stop job %ld. Insufficient permissions. (Not your job)\n", job.id);
                        }else
                        {
                            char time_buffer[15] = {0};
                            time_t now = time(NULL);
                            strftime(&time_buffer[0], 15, "%d-%m %H:%M:%S", localtime(&now));
                            fprintf(stderr, "[%s] User stopped job: %ld %ld\n", &time_buffer[0], w_node->job.id, (long) w_node->job.pid);
                            erase(m->waiting_queue, &w_node);
                            sprintf(&answer_buffer[0], GREEN "Success" RESET ": Stopped job %ld\n", job.id);
                        }
                        break;
                    }
                    w_node = w_node->next;
                }
            }
            pthread_mutex_unlock(m->waiting_lock);
            if( !found_job ){
                sprintf(&answer_buffer[0], YELLOW "Warning" RESET ": Job id %ld does not exist.\n", job.id);
            }
            send(temp_descriptor, answer_buffer, strlen(answer_buffer), 0);
        }
        if( buffer[0] == MSG_STATUS ){
            char answer_buffer[ANSWER_BUFFER] = {0};
            int n = 0;
            n += snprintf(&answer_buffer[n], ANSWER_BUFFER-n, "available cores: %ld\n", get_free_cores(m));
            n += snprintf(&answer_buffer[n], ANSWER_BUFFER-n, "job id\tstatus\t\tuser\t\tcores\tstart\t\tend\t\tcommand\n");
            pthread_mutex_lock(m->running_lock);
            struct Node* running_nodeent = m->running_queue->first;
            while( running_nodeent != NULL ){
                struct Job j = running_nodeent->job;
                char start_time[20] = {0};
                char end_time[20] = {0};
                strftime(&start_time[0], 15, "%d-%m %H:%M:%S", localtime(&j.start_time));
                strftime(&end_time[0], 15, "%d-%m %H:%M:%S", localtime(&j.end_time));
                n += snprintf(&answer_buffer[n], ANSWER_BUFFER-n,"%ld\t%s[running]%s\t%s\t%ld\t%s\t%s\t%s\n", (long)j.id, GREEN, RESET, j.user_name, j.cores, &start_time[0], &end_time[0], &j.cmd[0]);
                running_nodeent = running_nodeent->next;
            }
            pthread_mutex_unlock(m->running_lock);
            if( m->priority_job != NULL ){
                struct Job j = m->priority_job->job;
                char start_time[20] = {0};
                char end_time[20] = {0};
                strftime(&start_time[0], 15, "%d-%m %H:%M:%S", localtime(&j.start_time));
                strftime(&end_time[0], 15, "%d-%m %H:%M:%S", localtime(&j.end_time));
                n += snprintf(&answer_buffer[n], ANSWER_BUFFER - n, "%ld\t%s[priority]%s\t%s\t%ld\t%s\t%s\t%s\n", (long) j.id, CYAN, RESET, j.user_name, j.cores, &start_time[0], &end_time[0], &j.cmd[0]);
            }
            pthread_mutex_lock(m->waiting_lock);
            struct Node* waiting_node = m->waiting_queue->first;
            while( waiting_node != NULL ){
                struct Job j = waiting_node->job;
                n += snprintf(&answer_buffer[n], ANSWER_BUFFER - n, "%ld\t%s[waiting]%s\t%s\t%ld\t%s\t%s\t%s\n", (long) j.id, RED, RESET, j.user_name, j.cores, "n/a\t", "n/a\t", &j.cmd[0]);
                waiting_node = waiting_node->next;
            }
            pthread_mutex_unlock(m->waiting_lock);

            send(temp_descriptor, answer_buffer, strlen(answer_buffer), 0);
        }

        sleep(1);
    }
    return NULL;
}

util.c

just some helper functions

#include <errno.h>
#include <math.h>
#include <limits.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include "util.h"

int parse_long(const char* str, long* return_value){
    errno = 0;
    char* dummy;
    long value = strtol(str, &dummy, 0);

    if( dummy == str || *dummy != '\0' || ((value == LONG_MIN || value == LONG_MAX) && errno == ERANGE) ){
        return 0;
    }
    *return_value = value;
    return 1;
}

void quit(const char* msg){
    printf("%s", msg);
    exit(EXIT_FAILURE);
}

void quit_with_error(const char* msg){
    perror(msg);
    exit(EXIT_FAILURE);
}

char** split(char* str, char tok, int *num){
    int len = strlen(str);
    *num = 2;
    char* it = str;
    for( int i = 0; i < len; ++i ){
        if( *it == tok ){
            *it = '\0';
            (*num)++;
        }
        it++;
    }
    char** result = (char**)malloc(*num * sizeof(char*));
    int j = 0;
    int save = 1;
    it = str;
    for( int i = 0; i < len; ++i, ++it ){
        if( *it == '\0' ){
            save = 1;
        }
        else if( save == 1 ){
            result[j++] = it;
            save = 0;
        }
    }
    result[j] = (char*)NULL;
    return result;
}
````

Answer 1

Use the appropriate types

You use long and long long in several places. The C standard specifies the relative size between short/int/long/long long, but the exact size depends on the platform, and might be smaller or larger than you think. Make sure you choose the appropriate type. Here is a list of variables whose types should be changed:

• long port: a TCP port number is only 16 bits. A long is often way too big. Use uint16_t instead.
• long time_limit, long maxtime: on 64-bit Windows platforms and most 32-bit platforms, storing UNIX time in seconds in a long will cause it to wrap in 2038. Apart from that, it doesn't match the return type of time(). Be consistent and use time_t instead.
• unsigned long long core_mask: on most platforms this is only 64 bits, but this is not enough if you have an AMD 3990X CPU in your desktop PC, which has 128 logical cores, or if you are running on a multi-processor server which might have even more cores. On Linux, consider using cpu_set_t to store core masks.
• char server_ip[40]: prefer parsing any textual IP address before storing it in a struct. That avoids any issues with string lengths. 40 is way too much if you only support numeric IPv4 addresses, but if you want to support hostnames, it is too short. It might have been just enough for numeric IPv6 addresses, but it doesn't seem like you support IPv6 in your code. I would use struct sockaddr_storage, which stores both address and port, and supports IPv4 and IPv6.

You mentioned in the comments that you used long because that matches the return value of strtol(), which makes sense in a way. However, it is better to store values in the type associated with the intended use later on, as that will avoid implicit casts which might be problematic. It might also be a good idea to parse numbers into a temporary long variable, then check if the value is not too large (for example, a port number should never be larger than 65535) before storing it into the variable of the correct type.

Simplify zeroing structs

In prepare_submit_message(), you spend a lot of lines of code initializing job. You can simplify this a lot by using designated initializers:

struct Job job = {
    .user_id = geteuid(),
    .group_id = getegid(),
};

All the fields of struct Job that are not explicitly specified will be set to zero, so no more need for = 0 and memset().

Print all error messages to stderr

You are inconsistent in how you are printing error messages. Make sure all of them get printed to stderr. Ideally, usage information printed after an error parsing the command line should also go to stderr.

If you are targetting only Linux and/or BSD platforms, consider using err() and warn() from <err.h>, which combine printing a custom message, the result of strerror(errno), and if desired, quiting with a non-zero exit code all in one go. Or you can indeed use quit_with_error_message(), but then make sure you use it consistently.

Support IPv6

Your code only works for IPv4. We've run out of IPv4 addresses, so you should really make sure your code supports IPv6 as well. It's not much work: ensure you use functions that are independent of IP version to manipulate addresses, such as getaddrinfo() and getnameinfo(). getaddrinfo() will also return the socket type that as appropriate for a given address, so you can pass that to socket().

On Linux at least, if you bind to IPv6, and make sure the socket option IPV6_V6ONLY is set to false, it will also bind to IPv4, so you only need a single listening socket to support incoming connections from both IPv4 and IPv6 addresses. Otherwise, consider creating two listening sockets; one for IPv4 and one for IPv6.

Missing or incorrect error checking

If getcwd() returns NULL, you assume that the only possibe error could be that the working directory name is longer than your buffer. However, there are other possible reasons it can fail. Don't make assumptions, let perror() or err() print the actual error message.

send() and read() can succeed but send or receive less bytes than requested. This is especially likely for TCP sockets. Make sure you handle this scenario and send/receive the remaining bytes if this happens.

Sometimes you don't check return values at all. For example, you don't check the return value of accept().

There are also some functions that programmers typically don't check the return value of, because errors are very unlikely, but even snprintf(), pthread_mutex_lock(), strftime() and so on can return errors. Consider for all functions you call whether it can return an error, and if so what could go wrong if you don't handle errors correctly.

Don't quit due to network errors

Network errors are likely to happen, are not under your control, and temporary network issues can resolve themselves after a while. If something goes wrong when receiving data from a socket or sending data to a socket, don't quit your program, instead just close the socket and wait for the next one.

Clean up properly

If you accept() a socket, make sure you close() it after you are done with it. Failure to do so will cause resource leaks, and may cause your program to stop after running for a while because it ran out of memory or file descriptors.

If you call pthread_mutex_init(), make sure you call pthread_mutex_destroy() after you have finished using that mutex. Maybe it's not important since your program is going to exit anyway, but it helps static and runtime analyzers ensure you have no resource leaks.

Denial of service attack

It's easy to prevent anyone from submitting jobs to your job server: just open a TCP connection to its listening socket, and then don't send anything.

Don't sleep unnecessarily

There is a sleep(1) in server(). Why force a one second wait between incoming connections? It does not seem useful, and will only make job submissions take longer than necessary.

The calls to usleep() are also problematic. First, usleep() is a deprecated POSIX function, you should use nanosleep() or clock_nanosleep() instead. Apart from that, don't assume that sleeping for a bit after kill() is enough to ensure that the process you just sent SIGKILL are really killed. In any case, you either wait longer than necessary, or not long enough. Consider using waitpid() without WNOHANG to wait until a process is no longer running.

As for the sleep(1) inside queue(), you do want to sleep if there are no jobs to run, otherwise it would just spin and keep the CPU busy for no good reason. But again, just sleeping a random amount of time is not great, as it puts a limit on the number of jobs you can run in a given amount of time. In this case, use a condition variable that you signal inside server() whenever a new job is submitted, and which you wait for in queue(). For waking up if a job process has exited, you can register a handler for SIGCHILD

Don't call external programs unnecessarily

The setting of CPU affinities is ultimately done over taskset, although I feel I could have implemented that myself aswell, [...]

Starting an external program has lots of issues. Apart from the large amount of code you had to write, the nasty string manipulation you had to do, it is very inefficient. taskset is Linux-specific anyway, so since you don't have to worry about being platform independent, just use sched_setaffinity() to change the core affinity mask of a given process.

Missing mutex for available_cores

Several threads can modify available_cores, but no lock is taken to ensure the value is modified atomically. This means the core mask can potentially become corrupt, and either too many jobs will run simultaneously, or not all cores will be used. A simple solution is to add a mutex for it, or to use one of the existing mutexes to guard this variable.