Многопоточное программирование

Суть

Обычные программы – одна точка выполнение

Потоки – несколько точек выполнения. Вся память и ресурсы общие.

Зачем?

Использование ресурсов SMP-систем

Уменьшение времени реакции

Более естественная реализация

Альтернативы

Несколько программ

ls -l | awk '{print $5, $9}' | sort -n -r | head -n 20

Проблемы с передачей данных

Цикл событий

Требует модификации алгоритма

Пример: числа Фиббоначи

void fib(int* result)

{

int n1 = 1;

int n2 = 1;

while (n2 < 1000) {

int next = n1 + n2;

*result++ = next;

n1 = n2;

n2 = next;

}

}

Использование цикла событий

void fib(int& n1, int& n2,

int* result)

{

for (int i = 0; i < 10; ++i)

{

*result = n1 + n2;

n1 = n2

n2 = *result;

}

}

def fib():

n1 = 1

n2 = 1

while n2 < 1000:

result = n1 + n2

n1 = n2

n2 = result

yield result

Вывод

Потоки иногда нужны

Организация многопоточных программ

По потоку на запрос/задачу

Группа потоков-обработчиков

Проблемы

Проблемы

Нарушение последовательной логики

Недетерминизм времени

Средства синхронизации могут

заблокировать процесс

убить производительность

Пример: параллельная сортировка

Массив разбивается на две части, которые сортируются параллельно

Результаты сливаются

Пример использует библиотеку Boost.Threads:

http://boost.org/libs/thread

Рабочий поток

class Worker {

public:

Worker(int* start, int* end) : m_start(start), m_end(end) {}

void sort()

{

std::cout << "Worker started\n";

std::sort(m_start, m_end);

}

private:

int* m_start;

int* m_end;

};

Управляющая часть

void sort(int* start, int* end)

{

int* middle = start + (end-start)/2;

Worker w1(start, middle);

boost::thread t1(bind(&Worker::sort, ref(w1)));

Worker w2(middle, end);

boost::thread t2(bind(&Worker::sort, ref(w2)));

std::inplace_merge(start, middle, end);

}

main

int i[10] = {9,8,7,6,5,4,3,2,1,0};

sort(i, i + 10);

copy(i, i + 10, ostream_iterator<int>(cout, "\n"));

Ожидание

int finished = 0;

.....

void sort()

{

std::cout << "Worker started\n";

std::sort(m_start, m_end);

++finished;

}

.....

while(finished != 2);

std::inplace_merge(start, middle, end);

Взаимное исключение

WWorker started

orker started

Взаимное исключение

class Worker { ....

void sort()

{

boost::mutex::scoped_lock l(m);

std::cout << "Worker started\n";

std::sort(m_start, m_end);

++finished;

}

private:

boost::mutex m;

....

}

Правило 1

Все доступы к разделяемым переменным должны быть синхронизованы и использованием mutex.

Переменные синхронизации должны быть разделяемыми

Конфликты потоков

void sort()

{

boost::mutex::scoped_lock l(m);

std::cout << "Worker started\n";

std::sort(m_start, m_end);

boost::mutex::scoped_lock l2(finished_mutex);

++finished;

}

Конфликты потоков

void sort()

{

boost::mutex::scoped_lock l(m);

std::cout << "Worker started\n";

l.unlock();

std::sort(m_start, m_end);

boost::mutex::scoped_lock l2(finished_mutex);

++finished;

}

Проблема: активное ожидание

boost::mutex::scoped_lock l(finished_mutex);

while(finished != 2);

Эффективное ожидание

boost::mutex::scoped_lock l(finished_mutex);

while(finished != 2) {

l.unlock();

efficient_wait();

l.lock();

}

.....

++finished;

notify_all();

Condition variable

boost::condition finished_cond;

....

boost::mutex::scoped_lock l(finished_mutex);

while(finished != 2) {

finished_cond.wait(l);

}

......

boost::mutex_scoped_lock l(finished_mutex);

++finished;

finished_cond.notify_all();

Правило 2

Для ожидания используется сondition variable и цикл, проверяющий требуемое условие.

Цикл является обязательным.

Любое изменение разделяемых данных, которые могут использоваться в ожидании, заканчиватся вызовом notify_all

Высокоуровневые конструкции

Очередь

Барьер

RW-lock

Очередь

Несколько потоков пишут и читают

Если при чтении нет данных, поток ждет

Если при записи нет места, поток ждет

Использование блокировок

template<class T>

class Queue {

public:

bool empty() {

boost::mutex::scoped_lock l(m);

return m_wrapped.empty();

}

T& top() {

boost::mutex::scoped_lock l(m);

return m_wrapped.top();

}

private:

std::queue<T> m_wrapped;

boost::mutex m;

};

Race condition

Queue<Task*> queue;

if (!queue.empty())

Task* t = queue.top();

t->.....

Правильное решение

template<class T> сlass Queue {

boost::mutex m;

boost::condition c;

std::queue m_wrapped;

public:

T get() {

boost::mutex::scoped_lock l(m);

while(m_wrapped.empty()) c.wait(l);

T result = m_wrapped.top(); m_wrapped.pop();

return result;

}

void put(const T& t) {

boost::mutex::scoped_lock l(m);

m_wrapped.push(t);

c.notify_all();

}

};

Барьер

Средство синхронизации нескольких потоков

Поток может пройти барьер, когда все остальные потоки тоже подошли к барьеру

Барьер: прямая реализация

class barrier {

public:

barrier(int count) : m_count(count) {}

void wait()

{

boost::mutex::scoped_lock l(m);

if (--m_count == 0)

c.notify_all();

while(m_count != 0) {

c.wait(l);

}

}

private:

boost::mutex m;

boost::condition c;

};

Многоразовый барьер

void wait()

{

boost::mutex::scoped_lock l(m);

int initial_flag = flag;

if (--m_count == 0)

{

m_count = m_limit;

flag = (flag == 0) ? 1 : 0;

c.notify_all();

}

while(flag == initial_flag)

c.wait(l);

}

flags++;

RW-lock: мотивация

Много доступов к разделяемой структуре на чтение и мало на запись

Полная блокировка неэффективна

RW-lock

В любой момент любое количество потоков могут иметь доступ на чтение.

Только один поток поток может иметь доступ на запись

Одновременный доступ на чтение и запись запрещен

RW-lock: решение

class RW_lock {

public:

void read_lock()

{

boost::mutex::scoped_lock l(m);

while(m_writers > 0)

c.wait(l);

++m_readers;

}

void write_lock()

{

boost::mutex::scoped_lock l(m);

while(m_writers > 0 || m_readers > 0)

c.wait(l);

++m_writers;

}

.........

2004-02-18T11:49:46