futex.c

Shows how Linux futex syscall can be used to distribute computation to multiple threads.

/*
 * Futex demonstration program
 * [e-mail]
 */
#include <stdio.h>
#include <stdint.h>
#include <pthread.h>
#include <unistd.h>
#include <asm/unistd.h>
typedef uint32_t u32; /* work around broken futex.h */
#include <linux/futex.h>
#define futex(a,b,c,d) syscall(__NR_futex,a,b,c,d)

#define WORKER_THREADS 3

/* shared variables */
/* volatile */ int workers_active, ready_futex, run_futex, shutdown;
double input;
double partial_result[WORKER_THREADS];

double
eval(int thread)
{
        /* example evaluation function */
        return input * thread + thread;
}

void
await_work()
{
        int last_run_seq_no = run_futex;

        /* if we are the last thread to complete, signal ready_futex */
        if (__sync_fetch_and_add(&workers_active, -1) == 1)
        {
            ready_futex = 1;
            futex(&ready_futex, FUTEX_WAKE, 1, NULL);
        }

        /* now wait for run_futex to increment */
        if (run_futex == last_run_seq_no)
            futex(&run_futex, FUTEX_WAIT, last_run_seq_no, NULL);
}

void
thread_pool_wait()
{
        /* wait for ready_futex */
        if (!ready_futex)
            futex(&ready_futex, FUTEX_WAIT, 0, NULL);
}

void
thread_pool_notify()
{
        /* increment run_futex to start workers */
        workers_active = WORKER_THREADS;
        ready_futex = 0;
        run_futex++;
        futex(&run_futex, FUTEX_WAKE, WORKER_THREADS, NULL);
}

void *
worker(void *arg)
{
        int thread = (int)arg;

        printf("thread=%d\n", thread);
        while (1)
        {
            await_work();
            if (shutdown)
                return NULL;

            partial_result[thread] = eval(thread+1);
        }
}

void
evaluate(double arg)
{
        double result;
        int i;

        /* prepare inputs here */
        input = arg;

        /* start slaves */
        thread_pool_notify();
        /* might as well run an evaluation on the master too */
        result = eval(0);
        /* wait on any outstanding slaves */
        thread_pool_wait();

        /* accumulate results */
        for (i = 0; i < WORKER_THREADS; i++)
                result += partial_result[i];
        printf("result: %lf\n", result);
}

int
main()
{
        pthread_t threads[WORKER_THREADS];
        int i;

        workers_active = WORKER_THREADS;
        ready_futex = 0;
        shutdown = 0;
        for (i = 0; i < WORKER_THREADS; i++)
            pthread_create(&threads[i], NULL, worker, (void *)i);

        /* wait for threads to start up */
        thread_pool_wait();

        printf("starting...\n");
        evaluate(1);
        evaluate(2);
        evaluate(3);
        evaluate(4);
        evaluate(5);
        printf("done\n");

        /* signal threads to shut down */
        shutdown = 1;
        thread_pool_notify();

        for (i = 0; i < WORKER_THREADS; i++)
            pthread_join(threads[i], NULL);
        return 0;
}