scheduler.c

/*
 * SMT-Aware Multi-Process Job Scheduling System
 * Central Scheduler Process
 * 
 * The scheduler receives jobs from producers, maintains a priority queue,
 * and dispatches jobs to workers based on their load and SMT considerations.
 */

#include "common.h"

/* Priority queue node */
typedef struct pq_node {
    job_t job;
    struct pq_node *next;
} pq_node_t;

/* Priority queue */
typedef struct {
    pq_node_t *head;
    int size;
    pthread_mutex_t lock;
} priority_queue_t;

static volatile sig_atomic_t running = 1;
static int msg_queue_id;
static shared_state_t *shared_mem;
static int shm_id;
static priority_queue_t job_queue;

void signal_handler(int sig) {
    (void)sig;
    running = 0;
}

/* Initialize priority queue */
void pq_init(priority_queue_t *pq) {
    pq->head = NULL;
    pq->size = 0;
    pthread_mutex_init(&pq->lock, NULL);
}

/* Insert job into priority queue (higher priority first) */
void pq_insert(priority_queue_t *pq, job_t job) {
    pq_node_t *new_node = malloc(sizeof(pq_node_t));
    new_node->job = job;
    new_node->next = NULL;
    
    pthread_mutex_lock(&pq->lock);
    
    if (pq->head == NULL || job.priority > pq->head->job.priority) {
        new_node->next = pq->head;
        pq->head = new_node;
    } else {
        pq_node_t *current = pq->head;
        while (current->next != NULL && 
               current->next->job.priority >= job.priority) {
            current = current->next;
        }
        new_node->next = current->next;
        current->next = new_node;
    }
    
    pq->size++;
    pthread_mutex_unlock(&pq->lock);
}

/* Remove and return highest priority job */
int pq_pop(priority_queue_t *pq, job_t *job) {
    pthread_mutex_lock(&pq->lock);
    
    if (pq->head == NULL) {
        pthread_mutex_unlock(&pq->lock);
        return -1;
    }
    
    pq_node_t *node = pq->head;
    *job = node->job;
    pq->head = node->next;
    pq->size--;
    free(node);
    
    pthread_mutex_unlock(&pq->lock);
    return 0;
}

/* Select best worker based on load and SMT awareness */
int select_worker(void) {
    int best_worker = 0;
    int min_load = shared_mem->workers[0].current_load;
    
    for (int i = 1; i < MAX_WORKERS; i++) {
        int load = shared_mem->workers[i].current_load;
        
        /* SMT-aware selection: prefer workers with fewer active threads
         * to reduce contention on shared physical cores */
        if (load < min_load) {
            min_load = load;
            best_worker = i;
        }
        
        /* If loads are equal, prefer workers on different physical cores
         * Assuming 2 logical cores per physical core (SMT-2) */
        if (load == min_load && (i / 2) != (best_worker / 2)) {
            /* Different physical core - reduces SMT contention */
            best_worker = i;
        }
    }
    
    return best_worker;
}

/* Initialize shared memory and IPC */
int init_ipc(void) {
    /* Create message queue */
    msg_queue_id = msgget(MSG_QUEUE_KEY, IPC_CREAT | 0666);
    if (msg_queue_id == -1) {
        LOG_ERROR("Failed to create message queue: %s", strerror(errno));
        return -1;
    }
    
    /* Create shared memory */
    shm_id = shmget(SHM_KEY, sizeof(shared_state_t), IPC_CREAT | 0666);
    if (shm_id == -1) {
        LOG_ERROR("Failed to create shared memory: %s", strerror(errno));
        return -1;
    }
    
    /* Attach shared memory */
    shared_mem = shmat(shm_id, NULL, 0);
    if (shared_mem == (void *)-1) {
        LOG_ERROR("Failed to attach shared memory: %s", strerror(errno));
        return -1;
    }
    
    /* Initialize shared memory */
    memset(shared_mem, 0, sizeof(shared_state_t));
    pthread_mutexattr_t attr;
    pthread_mutexattr_init(&attr);
    pthread_mutexattr_setpshared(&attr, PTHREAD_PROCESS_SHARED);
    pthread_mutex_init(&shared_mem->global_lock, &attr);
    
    for (int i = 0; i < MAX_WORKERS; i++) {
        shared_mem->workers[i].worker_id = i;
        pthread_mutex_init(&shared_mem->workers[i].lock, &attr);
    }
    
    pthread_mutexattr_destroy(&attr);
    
    LOG_SCHEDULER("IPC initialized - MsgQ ID: %d, SHM ID: %d", 
                  msg_queue_id, shm_id);
    return 0;
}

/* Cleanup IPC resources */
void cleanup_ipc(void) {
    /* Signal shutdown to all processes */
    shared_mem->shutdown_flag = 1;
    
    /* Send shutdown messages to workers */
    for (int i = 0; i < MAX_WORKERS; i++) {
        job_message_t shutdown_msg;
        shutdown_msg.msg_type = MSG_TYPE_DISPATCH + i;
        shutdown_msg.target_worker = i;
        shutdown_msg.job.job_id = -1;  /* Shutdown signal */
        msgsnd(msg_queue_id, &shutdown_msg, sizeof(job_message_t) - sizeof(long), 0);
    }
    
    sleep(1);  /* Give workers time to shutdown */
    
    /* Detach and remove shared memory */
    shmdt(shared_mem);
    shmctl(shm_id, IPC_RMID, NULL);
    
    /* Remove message queue */
    msgctl(msg_queue_id, IPC_RMID, NULL);
    
    LOG_SCHEDULER("IPC resources cleaned up");
}

/* Dispatcher thread - sends jobs to workers */
void *dispatcher_thread(void *arg) {
    (void)arg;
    
    while (running) {
        job_t job;
        
        if (pq_pop(&job_queue, &job) == 0) {
            int target = select_worker();
            
            job_message_t msg;
            msg.msg_type = MSG_TYPE_DISPATCH + target;
            msg.target_worker = target;
            msg.job = job;
            
            if (msgsnd(msg_queue_id, &msg, sizeof(job_message_t) - sizeof(long), 0) == 0) {
                pthread_mutex_lock(&shared_mem->workers[target].lock);
                shared_mem->workers[target].current_load++;
                pthread_mutex_unlock(&shared_mem->workers[target].lock);
                
                LOG_SCHEDULER("Dispatched %s to Worker %d (load: %d)",
                             job.job_name, target, 
                             shared_mem->workers[target].current_load);
            }
        } else {
            usleep(10000);  /* Wait 10ms if queue empty */
        }
    }
    
    return NULL;
}

/* Stats printer thread */
void *stats_thread(void *arg) {
    (void)arg;
    
    while (running) {
        sleep(3);
        
        printf("\n" COLOR_BLUE "═══════════════ SCHEDULER STATS ═══════════════\n");
        printf("Queue Size: %d | Submitted: %d | Completed: %d\n",
               job_queue.size, 
               shared_mem->total_jobs_submitted,
               shared_mem->total_jobs_completed);
        
        printf("Worker Loads: ");
        for (int i = 0; i < MAX_WORKERS; i++) {
            printf("W%d[%d] ", i, shared_mem->workers[i].current_load);
        }
        printf("\n═════════════════════════════════════════════\n" COLOR_RESET);
    }
    
    return NULL;
}

int main(void) {
    pthread_t dispatcher, stats;
    
    /* Setup signal handlers */
    signal(SIGINT, signal_handler);
    signal(SIGTERM, signal_handler);
    
    LOG_SCHEDULER("Starting SMT-Aware Job Scheduler");
    
    /* Initialize */
    if (init_ipc() != 0) {
        return EXIT_FAILURE;
    }
    pq_init(&job_queue);
    
    /* Start dispatcher thread */
    pthread_create(&dispatcher, NULL, dispatcher_thread, NULL);
    pthread_create(&stats, NULL, stats_thread, NULL);
    
    LOG_SCHEDULER("Waiting for jobs from producers...");
    
    /* Main receive loop */
    while (running) {
        job_message_t msg;
        
        /* Non-blocking receive with timeout simulation */
        ssize_t ret = msgrcv(msg_queue_id, &msg, sizeof(job_message_t) - sizeof(long),
                             MSG_TYPE_JOB, IPC_NOWAIT);
        
        if (ret > 0) {
            shared_mem->total_jobs_submitted++;
            pq_insert(&job_queue, msg.job);
            
            LOG_SCHEDULER("Received: %s (priority %d) - Queue size: %d",
                         msg.job.job_name, msg.job.priority, job_queue.size);
        } else if (errno == ENOMSG) {
            usleep(50000);  /* Wait 50ms if no messages */
        }
        
        /* Check for completion messages */
        ret = msgrcv(msg_queue_id, &msg, sizeof(job_message_t) - sizeof(long),
                     MSG_TYPE_COMPLETE, IPC_NOWAIT);
        if (ret > 0) {
            shared_mem->total_jobs_completed++;
        }
    }
    
    LOG_SCHEDULER("Shutting down...");
    
    pthread_cancel(dispatcher);
    pthread_cancel(stats);
    pthread_join(dispatcher, NULL);
    pthread_join(stats, NULL);
    
    cleanup_ipc();
    
    LOG_SCHEDULER("Scheduler terminated");
    return EXIT_SUCCESS;
}