Простой пул потоков

Возможность использовать потоки — несомненный плюс современной разработки приложений. А вот для языка программирования JavaScript одной из бед, например, является то, что скрипт может выполняться только в одном потоке. Зачастую достоинства сопровождаются недостатками, так что и здесь основные сложности проявляются особенно ярко тогда, когда в новые потоки отправляются небольшие, но частые задания. При таком подходе поток создается непосредственно при поступлении задания и разрушается по завершении его обработки. Порождение и уничтожение потока приводят к определенным издержкам, нередко зависящим от операционной системы, поскольку обычно механизмы работы с потоками реализованы в ядре ОС.

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

Чтобы создать простой пул потоков на языке программирования C++, потребуется определить следующие «сущности»: рабочий поток, пул потоков, задание и очередь заданий. В пуле постоянно работают потоки, и они самостоятельно берут по одному заданию на выполнение. Количество потоков, их запуск и останов определяются в классе ThreadPool.

Допустим, требуется выполнить многопоточную обработку файлов. В таком случае задание будет содержать полное имя файла, а задачей потока будет открыть этот файл и каким-то способом обработать его содержимое (листинг 1).

Листинг 1
class Task {
protected:
string filename;
public:
Task() {
filename = “ ”;
}
Task(string _filename) {
filename = _filename;
}
void SetFilename(string _filename) {
filename = _filename;
}
};

Класс очереди заданий (TaskQueue) инкапсулирует в себе множество объектов класса задания (Task) и защищает их от одновременного доступа из различных потоков. Для простейшего разделения доступа можно использовать мьютексы (при программировании для Unix) или критические секции (в том случае, если целевой платформой служит Microsoft Windows). Для хранения заданий в очереди следует использовать контейнер Queue (очередь), т.е. задания добавляются «в голову», а забираются «из хвоста» (листинг 2). Такой метод хранения заданий на обработку с равными приоритетами можно считать классическим и достаточно честным, чтобы применять его в большинстве систем.

Листинг 2
class TaskQueue {
pthread_mutex_t queueMutex;
queue _taskQueue;
public:
TaskQueue() {
// Инициализируем мьютекс
}
~TaskQueue() {
// Освобождаем ресурсы
}
void addTask(Task* task) { /* Добавление задания */ }
Task* AcquireAndRemoveTask() {
/*

Достаем задание из очереди.
Если очередь пустая, то возвращаем NULL
*/
}
};

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

Листинг 3
class ThreadPool
{ vector _threads;
TaskQueue* _tqueue;
public:
ThreadPool() {}
ThreadPool(unsigned int numofthreads) {
// Инициализация пула с потоками
for (unsigned int i=0; i++; i <
numofthreads) {
addThread(new WorkerThread, true);
}
}
~ThreadPool() {
// Тут следует удалять все потоки,
//которыми владеет пул
}
// Добавление потоков в пул.
// thread — указатель на поток
// Owned — является ли пул владельцем потока
void addThread(WorkerThread* thread, bool
Owned=false) {
_threads.push_back(thread);
}
void setTaskQueue(TaskQueue* tqueue) {
_tqueue = tqueue;
}
void ProcessTasks() {
// 1) Запускаем потоки, которые
//обрабатывают задания
// 2) Ждем окончания выполнения
//всех потоков
}
};

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

Листинг 4
int main() {
// Создаем пул и очередь заданий
ThreadPool* my_pool = new ThreadPool(5);
// Количество потоков
TaskQueue* taskqueue = new TaskQueue();
// Добавляем задания
taskqueue->addTask(new Task(“filename1”));
taskqueue->addTask(new Task(“filename2”));
taskqueue->addTask(new Task(“filename3”));
// ....
taskqueue->addTask(new Task(“filenameN”));
// Запускаем обработку заданий
my_pool->setTaskQueue(taskqueue);
my_pool->ProcessTasks();
// Удаляем пул и очередь заданий
delete my_pool;
delete taskqueue;
return 0;
}

Минусы такого способа — усложнение реализации самого потока и необходимость синхронизации по крайней мере одного общего ресурса — очереди заданий.

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

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

А JavaScript также становится многопоточным. По крайней мере, «движок» JavaScript V8, включенный в состав браузера Chrome от Google, уже поддерживает многопоточность.