Thread pool

Пул потоков
Thread pool
Схема принципа работы пула потоков
Тип	Параллельного программирования
Назначение	Эффективное управление многопоточностью
Плюсы	Снижение накладных расходов на создание и уничтожение потоков
Минусы	При неоптимально великом или малом размере теряет эффективность
Родственные шаблоны	Пул объектов, Производитель-потребитель
Описан в Design Patterns	Нет

Thread pool (с англ. — «Пул потоков») — паттерн проектирования программного обеспечения, предназначенный для эффективного управления потоками исполнения при параллельной обработке задач. Вместо создания и удаления нового потока под каждую задачу, паттерн предлагает создание набора потоков, обрабатывающих потокобезопасную очередь.

Основы

Широкое распространение получили многопоточные процессорные архитектуры. Они позволяют обрабатывать множество задач одновременно, что значительно повышает эффективность обработки клиентских запросов. Но каждое создание и уничтожение потока чревато накладными расходами на системные вызовы. В архитектуре многопоточных приложений, в которой на каждую задачу создаётся новый поток при большом количестве задач добавляются расходы на выделение памяти под собственный стек и переключение контекста.

Пул потоков снижает эти накладные расходы, заметно увеличивая производительность в ряде сценариев. Вместо создания потока под каждую задачу, пул создаётся единовременно, и таким образом расходы на создание потоков сокращаются с количества задач до количества потоков. В частности пул потоков сокращает расходы во время исполнения под высокой нагрузкой.

Размер

Высокое значение для производительности и эффективности пула потоков имеет его размер. Неоптимально большой размер пула потоков приводит к простою части потоков и неоправданным расходам на их содержание. В то же время слишком малый размер пула нивелирует его преимущества вследствие необходимости создавать новые потоки во время работы.

Исследования показали, что количество потоков в пуле, определённое в прямой пропорциональной зависимости от количества ядер центрального процессора при игнорировании прочих показателей, является неоптимальным с точки зрения производительности. При определении оптимального размера пула следует учитывать в том числе и текущую нагрузку на сервер.

Примеры

Пример на Си

#include <stdio.h>
#include <stdbool.h>
#include <pthread.h>


#define THREADS 4
#define QUEUE_SIZE 8
#define TASKS 8


typedef struct {
        void (*function)(const char *);
        const char *arg;
} Task;

typedef struct {
        pthread_mutex_t lock;
        pthread_cond_t tasknotify;
        pthread_cond_t donenotify;
        pthread_t *threads;
        Task *queue;
        Task *taskptr;
        bool shutdown;
} ThreadPool;


void
worker(ThreadPool *pool)
{
        Task task;

        for (;;) {
                pthread_mutex_lock(&pool->lock);

                while (pool->taskptr == NULL) {
                        if (!pool->shutdown) {
                                pthread_cond_wait(&pool->tasknotify, &pool->lock);
                        } else {
                                pthread_mutex_unlock(&pool->lock);
                                pthread_exit(NULL);
                        }
                }

                task = *pool->taskptr;
                if (pool->taskptr == pool->queue)
                        pool->taskptr = NULL;
                else
                        pool->taskptr--;
                pthread_mutex_unlock(&pool->lock);
                task.function(task.arg);
                pthread_cond_signal(&pool->donenotify);
        }
}

bool
taskadd(ThreadPool *pool, void (*function)(const char *), const char *id)
{
        pthread_mutex_lock(&pool->lock);

        if (pool->taskptr == pool->queue + QUEUE_SIZE - 1 || pool->shutdown) {
                pthread_mutex_unlock(&pool->lock);
                return false;
        }

        if (pool->taskptr == NULL)
                pool->taskptr = pool->queue;
        else
                pool->taskptr++;

        pool->taskptr->function = function;
        pool->taskptr->arg = id;        

        pthread_cond_signal(&pool->tasknotify);
        pthread_mutex_unlock(&pool->lock);

        return true;
}

void
taskwork(const char *id) 
{
        printf("Задача «%s» выполняется потоком %ld\n", id, (long)pthread_self());
}

int
main(void)
{
        Task queue[QUEUE_SIZE];
        pthread_t threads[THREADS];

        ThreadPool pool = {
                .threads = threads,
                .queue = queue,
                .taskptr = NULL,
                .shutdown = false
        };

        pthread_mutex_init(&pool.lock, NULL);
        pthread_cond_init(&pool.tasknotify, NULL);
        pthread_cond_init(&pool.donenotify, NULL);

        for (int i = 0; i < THREADS; i++) {
                pthread_create(&pool.threads[i], NULL, (void*)worker, &pool);
        }

        const char *taskid[TASKS] = { "Один", "Два", "Три", "Четыре", "Пять", "Шесть", "Семь", "Восемь" };

        for (int i = 0; i < TASKS; i++)
                taskadd(&pool, taskwork, taskid[i]);

        pthread_mutex_lock(&pool.lock);
        while (pool.taskptr != NULL)
                pthread_cond_wait(&pool.donenotify, &pool.lock);

        pool.shutdown = true;
        pthread_cond_broadcast(&pool.tasknotify);
        pthread_mutex_unlock(&pool.lock);

        for (int i = 0; i < THREADS; i++)
                pthread_join(pool.threads[i], NULL);

        pthread_mutex_destroy(&pool.lock);
        pthread_cond_destroy(&pool.tasknotify);
        pthread_cond_destroy(&pool.donenotify);

        puts("Пул потоков остановлен");

        return 0;
}