work-stealing-scheduler/src/sched.c

#include "../includes/sched.h"

#include <errno.h>
#include <stdio.h>

static struct scheduler sched;

/* Lance une tâche de la pile */
void *
worker_routine(void *arg)
{
    struct scheduler *s = (struct scheduler *)arg;

    while(1) {
        pthread_mutex_lock(&s->mutex);

        if(s->exit == 1) {
            pthread_mutex_unlock(&s->mutex);
            printf("finalement..rien a faire, ciao\n");

            break;
        }

        // S'il on a rien à faire
        if(s->top == -1) {
            // printf("attend..\n");
            pthread_cond_wait(&s->cond, &s->mutex);
            pthread_mutex_unlock(&s->mutex);
            // printf("reveillé!\n");
            continue;
        }
        // printf("lancement tâche #%d\n", s->top);

        // Extrait la tâche de la pile
        taskfunc f = s->tasks[s->top].f;
        void *closure = s->tasks[s->top].closure;
        s->top--;
        pthread_mutex_unlock(&s->mutex);

        // Exécute la tâche
        f(closure, s);

        // Signale s'il n'y a plus rien à faire
        if(s->top == -1) {
            printf("va falloir partir\n");
            pthread_mutex_lock(&s->mutex);
            s->exit = 1;
            pthread_cond_broadcast(&s->cond);
            pthread_mutex_unlock(&s->mutex);
        }
    }

    return NULL;
}

int
sched_init(int nthreads, int qlen, taskfunc f, void *closure)
{
    if(nthreads == 0) {
        nthreads = sched_default_threads();
    }

    // TODO : Actuellement on n'utilises pas qlen
    // => On utilise une pile de taille fixe
    (void)qlen;

    sched.top = -1;
    sched.exit = 0;

    if(pthread_mutex_init(&sched.mutex, NULL) != 0) {
        fprintf(stderr, "Can't init mutex\n");
        return -1;
    }

    if(pthread_cond_init(&sched.cond, NULL) != 0) {
        fprintf(stderr, "Can't init condition variable\n");
        return -1;
    }

    pthread_t threads[nthreads];
    for(int i = 0; i < nthreads; ++i) {
        if(pthread_create(&threads[i], NULL, worker_routine, &sched) != 0) {
            fprintf(stderr, "Can't create threads\n");
            return -1;
        }
    }

    if(sched_spawn(f, closure, &sched) != 0) {
        fprintf(stderr, "Can't create the initial task\n");
        return -1;
    }

    for(int i = 0; i < nthreads; ++i) {
        if((pthread_join(threads[i], NULL) != 0)) {
            fprintf(stderr, "Can't wait the thread %d\n", i);
            return -1;
        }
    }

    return 1;
}

int
sched_spawn(taskfunc f, void *closure, struct scheduler *s)
{
    pthread_mutex_lock(&s->mutex);

    if(s->top + 1 >= MAX_TASKS) {
        pthread_mutex_unlock(&s->mutex);
        errno = EAGAIN;
        fprintf(stderr, "Stack is full\n");
        return -1;
    }

    s->tasks[++s->top] = (taskinfo){f, closure};

    pthread_cond_signal(&s->cond);
    pthread_mutex_unlock(&s->mutex);

    return 0;
}