2
\$\begingroup\$

Please review the following code for thread pool in C (I wrote it to implement a threaded web server for my personal project). I have been programming in C++ and Java but never did any serious programming in C.

threadpool.h

#ifndef THREAD_POOL_H
#define THREAD_POOL_H

#include <pthread.h>

#define MAX_THREAD 10

typedef struct {
    void *(*function) (void *);
} runnable_task;


typedef struct safe_queue_node {
    runnable_task *task;
    struct safe_queue_node *next;
} safe_queue_node;


typedef struct safe_queue{
    struct safe_queue_node *head;
    struct safe_queue_node *tail;
    pthread_mutex_t mutex;
} safe_queue;


typedef struct {
    pthread_t threads[MAX_THREAD];
    safe_queue *queue;
    int count;
    volatile int should_close;
} thread_pool;


runnable_task *
new_runnable_task(
        void *(*function) (void *));


thread_pool *
new_thread_pool();


void
free_thread_pool(
        thread_pool *pool);


void
add_task_to_pool(
        thread_pool *pool,
        runnable_task *task);


void
shutdown_thread_pool(
        thread_pool *pool);

#endif

threadpool.c

#include "threadpool.h"
#include <stdlib.h>

runnable_task *
new_runnable_task(
        void *(*function) (void *)) {
    runnable_task *task = malloc(sizeof(runnable_task));
    task->function = function;
    return task;
}


safe_queue_node *
_new_safe_queue_node(
        runnable_task *task) {
    safe_queue_node *node = malloc(sizeof(safe_queue_node));
    node->task = task;
    node->next = NULL;
    return node;
}


safe_queue *
_new_safe_queue() {
    safe_queue *queue = malloc(sizeof(safe_queue));
    pthread_mutex_init(&queue->mutex, NULL);
    queue->head = NULL;
    queue->tail = NULL;
    return queue;
}


void
_free_safe_queue(
        safe_queue *queue) {
    free(queue);
}


runnable_task *
_get_from_queue(
        safe_queue *queue) {
    runnable_task *task = NULL;

    pthread_mutex_lock(&queue->mutex);
    if (queue->head != NULL) {
        task = queue->head->task;
        queue->head = queue->head->next;
        if (queue->head == NULL)
            queue->tail = NULL;
    }
    pthread_mutex_unlock(&queue->mutex);

    return task;
}


void *
_run_task(
        void *arg) {
    thread_pool *pool = (thread_pool *) arg;
    while (!pool->should_close) {
        runnable_task *task = _get_from_queue(pool->queue);
        if (task == NULL)
            continue;
        task->function(NULL);
    }

    return NULL;
}


thread_pool *
new_thread_pool(
        int count) {
    thread_pool *pool = malloc(sizeof(thread_pool));
    pool->queue = _new_safe_queue();
    pool->count = count <= MAX_THREAD ? count : MAX_THREAD;
    pool->should_close = 0;
    for (int i = 0; i < pool->count; i++) {
        pthread_create(pool->threads + i, NULL, _run_task, pool);
    }

    return pool;
}


void
free_thread_pool(
        thread_pool *pool) {
    _free_safe_queue(pool->queue);
    free(pool);
}


void
_add_to_queue(
        safe_queue *queue,
        runnable_task *task) {
    pthread_mutex_lock(&queue->mutex);
    if (queue->head == NULL) {
        queue->head = queue->tail = _new_safe_queue_node(task);
    } else {
        queue->tail->next = _new_safe_queue_node(task);
        queue->tail = queue->tail->next;
    }
    pthread_mutex_unlock(&queue->mutex);
}


void
add_task_to_pool(
        thread_pool *pool,
        runnable_task *task) {
    _add_to_queue(pool->queue, task);
}


void
shutdown_thread_pool(
        thread_pool *pool) {
    while (pool->queue->head != NULL) ;

    pool->should_close = 1;

    for (int i = 0; i < pool->count; i++)
        pthread_join(pool->threads[i], NULL);
}
\$\endgroup\$
  • \$\begingroup\$ A single threaded server should be able to handle 10K requests. Multithreading is overkill. Don't make the mistake of each request is served by a single dedicated thread. Note: Most of the time on requests is spent waiting for IO to become ready it is not highly computational so you want to concentrate on making sure your workers don't wait for IO but move to the next request that has IO available to processes (in or out). \$\endgroup\$ – Martin York Dec 6 '19 at 18:26
  • \$\begingroup\$ see the C10K problem: en.wikipedia.org/wiki/C10k_problem \$\endgroup\$ – Martin York Dec 6 '19 at 18:27
  • \$\begingroup\$ This is why the highly efficient web server Node.js is a single threaded server. \$\endgroup\$ – Martin York Dec 6 '19 at 18:29
3
\$\begingroup\$
  • Busy waiting on an empty queue is doubleplusungood. Add a condition variable to the queue, let the threads to pthread_cond_wait on it, and pthread_cond_signal it when adding the task to the queue.

  • The functions you don't want to export (those with the leading underscores) shall be static.

  • sizeof(type) may lead to a double maintenance problem. Prefer sizeof(var), as in

    thread_pool *pool = malloc(sizeof(*pool));
    
|improve this answer|||||
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.