Subversion Repositories Code-Repo

/*
 * Thread pool test program.
 *
 * This program creates a thread pool with a configurable number
 * of threads and then submits a configurable number of tasks.
 * Each task reports which thread executed it.
 *
 * An associated Python script (check.py) parses the output and
 * checks that 'nthreads' tasks are executing in parallel.
 *
 * This program makes idealizing and simplifying assumptions.
 *
 * First, it assumes that there are enough physical CPUs available
 * so that in fact 'nthreads' threads can execute. This may not
 * be true on a heavily loaded machine; test results will not be 
 * reliable under these circumstances.
 *
 * Second, the specification does not require that all threads
 * have reached the point where they wait on the pool's condition
 * variable.  To increase the likelihood that this is the case,
 * the test script pauses for 1 second before submitting tasks.
 *
 * Written by G. Back for CS3214 Spring 2010.
 */
#include <assert.h>
#include <pthread.h>
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <unistd.h>
#include <sys/time.h>
#include <sys/resource.h>
#include <time.h>

#include "threadpool.h"

// helper function to count number of threads in this process
static int count_number_of_threads(void);

/* Data to be passed to callable. */
struct callable_data {
    int number;
};

/* 
 * A callable. 
 *
 * Returns a string that reports its number, which thread
 * executes the task, and the start and end time of the 
 * task's execution.
 */
static void *
callable_task(struct callable_data * callable)
{
    char buf[1024];

    struct timeval start, end;
    assert(gettimeofday(&start, NULL) == 0);

    struct timespec t = { .tv_sec = 0, .tv_nsec = 5000000 };
    assert(nanosleep(&t, NULL) == 0);

    assert(gettimeofday(&end, NULL) == 0);

    snprintf(buf, sizeof buf, 
            "Future #%d Thread #%p start=%ld.%ld end=%ld.%ld", 
             callable->number, (void *)pthread_self(),
            start.tv_sec, start.tv_usec,
            end.tv_sec, end.tv_usec);
    return strdup(buf);
}

int
main(int ac, char *av[])
{
    assert (ac > 2 || !!!"Usage: threadpool_test <nthreads> <ntasks>");

    int nthreads = atoi(av[1]);
    int ntasks = atoi(av[2]);
    struct thread_pool * ex = thread_pool_new(nthreads);
    const int N = ntasks;
    struct future * f[N];

    // sleep .5 seconds to give threads time to start up
    struct timespec sleep_for = { .tv_sec = 0, .tv_nsec = 5*1e8 };
    nanosleep(&sleep_for, NULL);

    int threadsstarted = count_number_of_threads() - 1;
    if (threadsstarted != nthreads) {
        printf("The thread pool started %d instead of %d threads.\n",
                threadsstarted, nthreads);
        return EXIT_FAILURE;
    }

    // check for busy-waiting implementation
    struct rusage usage;
    int rc = getrusage(RUSAGE_SELF, &usage);
    if (rc == -1)
        perror("getrusage"), exit(-1);

    if (usage.ru_utime.tv_sec > 0 || usage.ru_utime.tv_usec > 400000) {
        printf("Thread pool is consuming excessive CPU time without running any jobs\n");
        return EXIT_FAILURE;
    }

    // submit N tasks and record futures obtained in return
    int i;
    for (i = 0; i < N; i++) {
        struct callable_data * callable_data = malloc(sizeof *callable_data);
        callable_data->number = i;
        f[i] = thread_pool_submit(ex, 
                               (thread_pool_callable_func_t) callable_task, 
                               callable_data);
    }
    printf("Main thread: %p\n", (void *)pthread_self());

    // wait for each future
    for (i = 0; i < N; i++) {
        printf("%s\n", (char *) future_get(f[i]));
    }

    // check that no pool thread shut down prematurely
    threadsstarted = count_number_of_threads() - 1;
    if (threadsstarted != nthreads) {
        printf("Only %d thread pool threads are left, should be %d threads.\n",
                threadsstarted, nthreads);
        return EXIT_FAILURE;
    }
    thread_pool_shutdown(ex);

    // sleep .3 seconds to give threads time to shut down
    // pthread_join() is not atomic with respect to the number
    // of tasks reported by /proc/self/status
    sleep_for.tv_nsec = 3*1e8;
    nanosleep(&sleep_for, NULL);

    int threadsleft = count_number_of_threads();
    if (threadsleft != 1) {
        printf("The thread pool did not correctly shut down"
               ", there are %d threads left.\n", threadsleft);
        return EXIT_FAILURE;
    }

    printf("Done.\n");
    return EXIT_SUCCESS;
}

/**
 * Count number of threads by scanning /proc/self/status
 * for the Threads: ... line
 */
static int
count_number_of_threads(void)
{
    FILE * p = fopen("/proc/self/status", "r");
    while (!feof(p)) {
        int threadsleft;
        char buf[128];
        fgets(buf, sizeof buf, p);
        if (sscanf(buf, "Threads: %d\n", &threadsleft) != 1)
            continue;

        fclose(p);
        return threadsleft;
    }
    printf("Internal error, please send email to gback@cs.vt.edu\n");
    abort();
}