This repository has been archived on 2024-05-05. You can view files and clone it, but cannot push or open issues or pull requests.
work-stealing-scheduler/src/sched.c

307 lines
7.6 KiB
C
Raw Normal View History

#include "../includes/sched.h"
2024-03-09 18:47:40 +01:00
2024-04-18 17:41:41 +02:00
#include <errno.h>
#include <pthread.h>
2024-03-15 12:43:44 +01:00
#include <stdio.h>
#include <stdlib.h>
2024-04-21 15:00:45 +02:00
#include <string.h>
2024-03-15 12:43:44 +01:00
struct task_info {
void *closure;
taskfunc f;
};
2024-04-18 17:41:41 +02:00
struct scheduler {
2024-04-21 15:03:00 +02:00
/* Taille des piles */
int qlen;
2024-04-19 15:33:03 +02:00
2024-04-22 02:09:51 +02:00
/* Variable de conditions pour reveillé les threads au besoin */
pthread_cond_t cond;
2024-04-21 15:02:31 +02:00
/* Mutex qui protège les piles */
2024-04-22 02:09:51 +02:00
pthread_mutex_t mutex;
2024-04-19 15:33:03 +02:00
2024-04-19 16:24:54 +02:00
/* Nombre de threads instanciés */
int nthreads;
2024-04-21 15:02:31 +02:00
/* Piles de tâches */
2024-04-21 15:00:45 +02:00
struct task_info **tasks;
2024-04-19 15:33:03 +02:00
/* Liste des threads */
pthread_t *threads;
2024-04-22 02:09:51 +02:00
/* Compteur des threads dormants */
int nthsleep;
2024-04-21 20:15:02 +02:00
2024-04-22 02:09:51 +02:00
/* Stack sous forme de dequeu pour gérer la récupération
* du premier élément ajouté */
int *top;
int *bottom;
};
2024-04-19 15:33:03 +02:00
/* Ordonnanceur partagé */
static struct scheduler sched;
2024-04-19 12:57:21 +02:00
/* Lance une tâche de la pile */
2024-04-19 16:27:58 +02:00
void *sched_worker(void *);
2024-04-19 15:33:03 +02:00
/* Nettoie les opérations effectuées par l'initialisation de l'ordonnanceur */
2024-04-21 14:42:36 +02:00
int sched_init_cleanup(int);
2024-04-21 16:39:00 +02:00
/* Récupère l'index du thread courant */
2024-04-21 16:51:26 +02:00
int current_thread(struct scheduler *);
2024-04-21 16:39:00 +02:00
2024-04-19 15:33:03 +02:00
int
sched_init(int nthreads, int qlen, taskfunc f, void *closure)
{
sched.tasks = NULL;
sched.threads = NULL;
2024-04-21 15:00:45 +02:00
sched.top = NULL;
2024-04-22 02:09:51 +02:00
sched.bottom = NULL;
if(qlen <= 0) {
fprintf(stderr, "qlen must be greater than 0\n");
return -1;
2024-04-18 17:41:41 +02:00
}
2024-04-22 02:09:51 +02:00
sched.qlen = qlen + 1; // circular buffer
2024-04-18 17:41:41 +02:00
if(nthreads < 0) {
fprintf(stderr, "nthreads must be greater than 0\n");
return -1;
} else if(nthreads == 0) {
nthreads = sched_default_threads();
}
2024-04-22 02:35:14 +02:00
sched.nthreads = 0;
2024-04-18 17:41:41 +02:00
2024-04-22 02:09:51 +02:00
sched.nthsleep = 0;
2024-04-21 20:15:02 +02:00
2024-04-22 02:09:51 +02:00
// Initialisation du mutex
if(pthread_mutex_init(&sched.mutex, NULL) != 0) {
fprintf(stderr, "Can't init mutex\n");
2024-04-21 14:42:36 +02:00
return sched_init_cleanup(-1);
}
2024-04-22 02:09:51 +02:00
// Initialisation variable de condition
if(pthread_cond_init(&sched.cond, NULL) != 0) {
fprintf(stderr, "Can't init varcond\n");
return sched_init_cleanup(-1);
2024-04-19 13:48:50 +02:00
}
2024-04-18 17:41:41 +02:00
2024-04-21 14:55:36 +02:00
// Initialisation du curseur suivant l'état de la pile de chaque processus
2024-04-22 02:35:14 +02:00
if(!(sched.top = malloc(nthreads * sizeof(int)))) {
2024-04-22 02:09:51 +02:00
perror("Cursor top stack");
return sched_init_cleanup(-1);
}
2024-04-22 02:35:14 +02:00
if(!(sched.bottom = malloc(nthreads * sizeof(int)))) {
2024-04-22 02:09:51 +02:00
perror("Cursor bottom stack");
2024-04-21 14:55:36 +02:00
return sched_init_cleanup(-1);
}
2024-04-22 02:35:14 +02:00
for(int i = 0; i < nthreads; ++i) {
2024-04-22 02:09:51 +02:00
sched.top[i] = 0;
sched.bottom[i] = 0;
2024-04-21 14:55:36 +02:00
}
2024-04-21 14:42:36 +02:00
2024-04-21 15:00:45 +02:00
// Allocation mémoire pour la pile de chaque processus
2024-04-22 02:35:14 +02:00
if(!(sched.tasks = malloc(nthreads * sizeof(struct task_info *)))) {
2024-04-21 15:00:45 +02:00
perror("Stack list");
2024-04-21 14:42:36 +02:00
return sched_init_cleanup(-1);
}
2024-04-22 02:35:14 +02:00
for(int i = 0; i < nthreads; ++i) {
2024-04-21 15:00:45 +02:00
if(!(sched.tasks[i] = malloc(qlen * sizeof(struct task_info)))) {
fprintf(stderr, "Stack for thread %d: %s\n", i, strerror(errno));
return sched_init_cleanup(-1);
}
}
2024-04-22 02:35:14 +02:00
// Initialise l'aléatoire
srand(time(NULL));
2024-04-21 19:29:21 +02:00
// Créer les threads
2024-04-22 02:35:14 +02:00
if(!(sched.threads = malloc(nthreads * sizeof(pthread_t)))) {
perror("Threads");
return sched_init_cleanup(-1);
}
2024-04-22 02:09:51 +02:00
// Ajoute la tâche initiale
if(sched_spawn(f, closure, &sched) < 0) {
fprintf(stderr, "Can't create the initial task\n");
return sched_init_cleanup(-1);
}
// Démarre les threads
2024-04-19 15:33:03 +02:00
for(int i = 0; i < nthreads; ++i) {
if(pthread_create(&sched.threads[i], NULL, sched_worker, &sched) != 0) {
2024-04-19 21:59:09 +02:00
fprintf(stderr, "Can't create the thread %d\n", i);
if(i > 0) {
fprintf(stderr, ", cancelling already created threads...\n");
for(int j = 0; j < i; ++j) {
if(pthread_cancel(sched.threads[j]) != 0) {
fprintf(stderr, "Can't cancel the thread %d\n", j);
}
}
} else {
fprintf(stderr, "\n");
}
2024-04-21 14:42:36 +02:00
return sched_init_cleanup(-1);
2024-04-19 15:33:03 +02:00
}
pthread_mutex_lock(&sched.mutex);
2024-04-22 02:35:14 +02:00
sched.nthreads++;
pthread_mutex_unlock(&sched.mutex);
2024-04-18 17:41:41 +02:00
}
2024-04-19 15:33:03 +02:00
for(int i = 0; i < nthreads; ++i) {
if((pthread_join(sched.threads[i], NULL) != 0)) {
2024-04-19 15:33:03 +02:00
fprintf(stderr, "Can't wait the thread %d\n", i);
2024-04-21 14:42:36 +02:00
return sched_init_cleanup(-1);
2024-04-19 15:33:03 +02:00
}
2024-04-18 17:41:41 +02:00
}
2024-04-21 14:42:36 +02:00
return sched_init_cleanup(1);
}
int
2024-04-21 14:42:36 +02:00
sched_init_cleanup(int ret_code)
{
2024-04-22 02:09:51 +02:00
pthread_mutex_destroy(&sched.mutex);
2024-04-21 14:42:36 +02:00
2024-04-22 02:09:51 +02:00
pthread_cond_destroy(&sched.cond);
2024-04-21 14:42:36 +02:00
if(sched.tasks) {
2024-04-21 15:00:45 +02:00
for(int i = 0; i < sched.nthreads; ++i) {
if(sched.tasks[i]) {
free(sched.tasks[i]);
sched.tasks[i] = NULL;
}
}
free(sched.tasks);
sched.tasks = NULL;
}
2024-04-19 21:16:04 +02:00
if(sched.threads) {
free(sched.threads);
sched.threads = NULL;
}
2024-04-21 14:55:36 +02:00
if(sched.top) {
free(sched.top);
sched.top = NULL;
}
2024-04-22 02:09:51 +02:00
if(sched.bottom) {
free(sched.bottom);
sched.bottom = NULL;
2024-04-21 16:51:26 +02:00
}
2024-04-22 02:09:51 +02:00
return ret_code;
2024-04-21 16:39:00 +02:00
}
2024-04-21 15:38:06 +02:00
2024-04-21 16:39:00 +02:00
int
2024-04-21 16:51:26 +02:00
current_thread(struct scheduler *s)
2024-04-21 16:39:00 +02:00
{
2024-04-21 16:51:26 +02:00
pthread_t current = pthread_self();
pthread_mutex_lock(&s->mutex);
2024-04-23 11:05:57 +02:00
for(int i = 0; i < s->nthreads; ++i) {
2024-04-21 16:51:26 +02:00
if(pthread_equal(s->threads[i], current)) {
pthread_mutex_unlock(&s->mutex);
2024-04-21 16:51:26 +02:00
return i;
}
}
pthread_mutex_unlock(&s->mutex);
2024-04-21 16:51:26 +02:00
return -1;
2024-04-21 15:38:06 +02:00
}
int
2024-04-22 02:09:51 +02:00
sched_spawn(taskfunc f, void *closure, struct scheduler *s)
2024-04-21 15:38:06 +02:00
{
2024-04-22 02:09:51 +02:00
int th;
if((th = current_thread(s)) < 0) {
th = 0;
}
2024-04-18 17:41:41 +02:00
2024-04-22 02:09:51 +02:00
pthread_mutex_lock(&s->mutex);
int next = (s->top[th] + 1) % s->qlen;
if(next == s->bottom[th]) {
pthread_mutex_unlock(&s->mutex);
2024-04-19 15:33:03 +02:00
fprintf(stderr, "Stack is full\n");
2024-04-22 02:09:51 +02:00
errno = EAGAIN;
2024-04-19 15:33:03 +02:00
return -1;
}
2024-04-18 17:41:41 +02:00
2024-04-22 02:09:51 +02:00
s->tasks[th][s->top[th]] = (struct task_info){closure, f};
s->top[th] = next;
2024-04-22 02:09:51 +02:00
pthread_mutex_unlock(&s->mutex);
2024-04-18 17:41:41 +02:00
2024-04-19 15:33:03 +02:00
return 0;
2024-03-09 18:47:40 +01:00
}
void *
2024-04-19 16:27:58 +02:00
sched_worker(void *arg)
{
struct scheduler *s = (struct scheduler *)arg;
2024-04-21 16:51:26 +02:00
// Récupère le processus courant (index tableau)
2024-04-21 18:04:41 +02:00
int curr_th;
2024-04-22 02:35:14 +02:00
while((curr_th = current_thread(s)) < 0);
2024-04-21 16:51:26 +02:00
2024-04-22 02:09:51 +02:00
struct task_info task;
int found;
while(1) {
2024-04-22 02:09:51 +02:00
found = 0;
pthread_mutex_lock(&s->mutex);
2024-04-21 18:04:41 +02:00
2024-04-22 02:09:51 +02:00
if(s->bottom[curr_th] != s->top[curr_th]) {
found = 1;
s->top[curr_th] = (s->top[curr_th] - 1 + s->qlen) % s->qlen;
task = s->tasks[curr_th][s->top[curr_th]];
}
2024-04-21 18:04:41 +02:00
2024-04-22 02:09:51 +02:00
if(!found) {
// Vol car aucune tâche trouvée
for(int i = 0, k = rand() % (s->nthreads + 1), target;
i < s->nthreads; ++i) {
2024-04-22 02:11:35 +02:00
target = (i + k) % s->nthreads;
2024-04-22 02:09:51 +02:00
if(s->bottom[target] != s->top[target]) {
// Tâche trouvée
found = 1;
s->top[target] = (s->top[target] - 1 + s->qlen) % s->qlen;
task = s->tasks[target][s->top[target]];
break;
2024-04-21 18:04:41 +02:00
}
2024-04-21 19:51:50 +02:00
}
2024-04-21 18:04:41 +02:00
2024-04-22 02:09:51 +02:00
// Aucune tâche à faire
if(!found) {
s->nthsleep++;
// Ne partir que si tout le monde dort
if(s->nthsleep >= s->nthreads) {
pthread_cond_broadcast(&s->cond);
pthread_mutex_unlock(&s->mutex);
2024-04-21 18:04:41 +02:00
2024-04-22 02:09:51 +02:00
break;
}
pthread_cond_wait(&s->cond, &s->mutex);
s->nthsleep--;
pthread_mutex_unlock(&s->mutex);
2024-04-21 15:38:06 +02:00
continue;
}
}
2024-04-22 02:09:51 +02:00
pthread_cond_broadcast(&s->cond);
pthread_mutex_unlock(&s->mutex);
// Exécute la tâche
2024-04-22 02:09:51 +02:00
task.f(task.closure, s);
}
return NULL;
}