Многопоточное программирование. 3. Проблема производителя и потребителя

Проблема производителя и потребителя, известная так же как проблема ограниченного буфера, является одной из классических задач многопроцессной синхронизации.

Рассмотрим упрощенный случай с двумя процессами. Первый процесс - Производитель (Producer), второй - Потребитель (Consumer).
Процессы используют общий буфер: Производитель помещает информацию в буфер, в то время как Потребитель извлекает ее, независимо от Производителя. Для Производителя всегда должно присутствовать незанятое место (слот) для записи информации, тогда как для Потребителя информация для извлечения должна присутствовать всегда, т.е. как минимум один непустой слот.
Производитель надеется на Потребителя, что тот предоставит свободное место для вставки очередной порции информации, а Потребитель надеется на Производителя, что тот обеспечит информацию для извлечения.
В таком случае Производитель и Потребитель каким-то образом должны взаимодействовать между собой, чтобы знать, что очередная вставка/извлечение будет безопасной для работающего в данный момент одного из процессов.

На первый взгляд решение, представленное ниже, абсолютно верное:

#define MAX_SIZE 100
int count = 0;
Producer()
{
  while (TRUE)
  {
    item = ProduceItem();
    if (count == MAX_SIZE) // (1) если буфер полон
    sleep(); // (2)
    InsertItem(item);
    ++count;
    if (count == 1)// был ли буфер пуст
      wakeup(Consumer);
  }
}

Consumer()
{
  while(TRUE)
  {
    if (count == 0) // (3) если буфер пуст
      sleep(); // (4)
    item = RemoveItem();
    --count;
    if (count == MAX SIZE - 1) // был ли буфер полон
      wakeup(Producer);
    ConsumeItem(item);
  }
}

count - количество слотов информации (или товаров, кому как) в буфере.
MAX_SIZE - максимальный размер буфера
Производитель сначала проверяет, не полон ли буфер, и если буфер полон, то Производитель засыпает, если не полон, то помещает информацию, увеличивает счетчик занятых слотов, и проверяет, не был ли буфер пусто до этого, если был пуст - значит Потребитель спит, и его надо разбудить.
Потребитель же наоборот, засыпает, когда буфер оказывается пустым. Если же после извлечения очередного элемента, Потребитель узнает, что до этого буфер был полным, значит Производитель спит и его надо разбудить.
Но в жизни все не так :-)

Представим, что буфер полон и работает Производитель. Осуществляется проверка на полноту буфера, и внезапно после выполнения строки (1), т.е. когда стало уже известно что надо заснуть, произошло переключение процесса. Теперь работает Потребитель. Он исправно извлекает информацию. Очистив первый слот, Потребитель понимает что буфер был полон, и нужно разбудет спящего, как ему кажется, производителя. Потребитель посылает сигнал пробуждения, и очищает буфер до конца, после чего засыпает. После этого происходит переключение на процесс Производителя, который не спал, а только собирался, и теперь он готов осуществить свою мечту - немного поспать. И засыпает. Все спят. Взаимоблокировка или deadlock.
А проблема, как уже многие догадались, состоит в утраченном вызове активизации процесса. Т.е. Производитель не спал, а его будили. К аналогичному результату привело бы рассмотрение ситуации, если бы буфер оказался пустым, и Потребитель решил бы уснуть, но "не успел".

Итак. Попробуем устранить проблему утраченного сигнала. Для этого будем использовать семафоры.
Cемафор - переменная, как правило целочисленная, используется для синхронизации доступа множества параллельно работающих процессов/потоков к общему ресурсу. Между семафором и обыкновенной численной переменной есть различия:

• Будучи созданным, семафору можно присвоить любое значение (для counting semaphore), но операции, которые могут работать с семафором, ограничиваются лишь инкрементом и декрементом
• Когда процесс/поток уменьшает значение семафора, и результат становится отрицательным, то он блокирует сам себя, и не может продолжать выполнение, пока другой процесс/поток не увеличит значение семафора
• Если процесс/поток инкрементирует значение семафора, и есть ожидающие процессы, то один из них разблокируется

В нашем решении мы будем использовать 2 вида семафоров:

• Counting semaphore (подсчитывающий)
• Binary semaphore (двоичный, или мьютекс)

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

#define MAX_SLOTS 100
typedef int semaphore;
semaphore mutex = 1;
semaphore empty = N; // подсчет пустых слотов в буфере
semaphore full = 0; // подсчет заполненных слотов буфере
Producer()
{
  int item;
  while (TRUE)
  {
    item = ProduceItem();
    down(empty); // (1)
    down(mutex); // (2)
    EnterItem(item);
    up(mutex); // (3)
    up(full); // (4)
  }
}
Consumer()
{
  int item;
  while (TRUE)
  {
    down(full); // (5)
    down(mutex); // (6)
    item = RemoveItem();
    up(mutex); // (7)
    up(empty); // (8)
    ConsumeItem(item);
  }
}

Итак, mutex отвечает за доступ одного процесса/потока в данный момент времени. Разберемся с empty и full. Пусть какое-то время программа работает, затем буфер заполняется полностью. Пусть в это время работает Производитель. Если буфер заполнен полностью, то empty = 0, это значит, что команда down(empty) приведет к блокировке Производителя (вспоминаем пункт 2 о семафорах). Далее работает Потребитель, разгружает слоты, регулирует семафоры количества пустых и полных слотов (количество пустых растет), и после первого очищения слота, Производитель может быть разблокирован, т.к. значение семафора увеличилось. Допустим произошло переключени на Производителя. Он продолжает выполнение со строки (2), заходит в критическую секцию, заполняет слот, выходит и увеличивает количество заполненных ячеек, тем самым позволяя Потребителю (если тот заблокирован) разблокироваться. В том и разница, что в данном случае Производитель не производит насильную разблокировку Потребителю, не задумываясь, надо ли это ему, а лишь создает условия, необходимые для разблокировки в случае, если таковая потребуется. Читателю предлагается проверить оставшиеся три граничные ситуации:

• Буфер полон, работает Потребитель
• Буфер пуст, работает Производитель
• Буфер пуст, работает Потребитель

Уфффф!