مسئله تولیدکننده-مصرف‌کننده

مسئله تولیدکننده-مصرف‌کننده یا مسئله بافر محدود مسئله‌ای کلاسیک در همگام‌سازی چندپروسه‌ای است که این پروسه‌ها یک ناحیه بحرانی مشترک دارند.
تشریح
در این مسئله، دو فرایند به نام‌های تولیدکننده و مصرف‌کننده وجود دارد که یک بافر با اندازه‌ای ثابت در بین آنها به اشتراک گذاشته شده و آنها از این بافر به عنوان صف استفاده می‌کنند. وظیفه تولیدکننده این است که یک قلم داده تولید کرده و در بافر قرار دهد. به شکل همزمان، مصرف‌کننده باید این قلم داده را از بافر بردارد و آن را به مصرف برساند. تولیدکننده و مصرف‌کننده تنها می‌توانند هر بار یک قلم داده را در بافر قرار دهند یا از بافر بردارند. مسئله اینجاست که ما باید مطمئن شویم که تولیدکننده سعی نمی‌کند در یک بافر پر شده، داده‌ای را ذخیره کند یا به طور مشابه، مصرف‌کننده سعی نکند داده‌ای را از یک بافر خالی بردارد. هرگاه که بافر پر بود و تولیدکننده خواست چیزی را در آن ذخیره کند، باید به خواب برود تا اینکه مصرف‌کننده یک قلم داده را از بافر بردارد و بعد از آن تولیدکننده را از خواب بیدار کند. به طور مشابه، اگر مصرف‌کننده خواست یک قلم داده را از یک بافر خالی بردارد، باید به خواب برود تا تولیدکنده یک قلم داده را در بافر قرار دهد و سپس مصرف‌کننده را بیدار کند. یک راه حل نامناسب برای این مسئله، می‌تواند به بن‌بست منجر شود که در آن هر دو پروس همزمان به خواب می‌روند و هیچکس نیست تا آنها را از خواب بیدار کند که باید جلوی این اتفاق را گرفت. این مسئله را می‌توان به n تولیدکننده و m مصرف‌کننده تعمیم داد.
پیاده‌سازی‌ها
پیاده‌سازی نادرست
برای حل این مشکل، ممکن است یک برنامه‌نویس ناوارد راه حل زیر را ارائه دهد. در این راه حل، دو رویه کتابخانه‌ای به نام sleep و wakeup مورد استفاده قرار گرفته‌اند. وقتی که فرایندی رویه sleep را فراخوانی می‌کند، فرایند فراخوان به خواب رفته تا زمانی که فرایند دیگر آن را توسط رویه wakeup از خواب بیدار کند. متغیر سراسری itemCount تعداد اقلام موجود در بافر را نشان می‌دهد.
int itemCount = 0;
procedure producer() {
while (true) {
item = produceItem();
if (itemCount == BUFFER_SIZE) {
sleep();
putItemIntoBuffer(item);
itemCount = itemCount + 1;
if (itemCount == 1) {
wakeup(consumer);
procedure consumer() {
while (true) {
if (itemCount == 0) {
sleep();
item = removeItemFromBuffer();
itemCount = itemCount - 1;
if (itemCount == BUFFER_SIZE - 1) {
wakeup(producer);
consumeItem(item);
مشکلی که در این راه حل وجود دارد این است که این راه حل یک وضعیت رقابتی دارد که می‌تواند منجر به پدیده بن‌بست شود. سناریو زیر را در نظر بگیرید:
مصرف‌کننده متغیر itemCount را می‌خواند و متوجه می‌شود که مقدار آن صفر است. این اتفاق دقیقا بعد از وارد شدن به بلاک if رخ می‌دهد.
درست قبل از فراخوانی رویه sleep، اجرای مصرف‌کننده توسط پردازنده متوقف شده و اجرای تولیدکننده آغاز می‌شود. (مثلا در یک سیستم اشتراک زمانی که پردازنده با سرعت زیاد بین فرایندها سوئیچ می‌کند)
تولیدکننده یک قلم داده را تولید کرده و آن را در بافر قرار داده و سپس یک واحد به itemCount اضافه می‌کند.
از آنجا که این متغیر قبلا صفر بوده است، تولیدکننده سعی می‌کند که مصرف‌کننده را از خواب بیدار کند.
متاسفانه مصرف‌کننده هنوز نخوابیده است و این وسط سیگنال wakeup گم می‌شود. وقتی که اجرای مصرف‌کننده ادامه می‌یابد، مصرف کننده بلافاصله به خواب رفته و دیگر هرگز بیدار نخواهد شد. به این خاطر که تولیدکننده تنها وقتی مصرف‌کننده را بیدار می‌کند که مقدار متغیر itemCount برابر ۱ باشد.
تولیدکننده آنقدر داده در بافر ذخیره می‌کند تا بافر پر شود. سپس خودش هم به خواب می‌رود. اما مصرف‌کننده‌ای نیست که آن را از خواب بیدار کند.
از آنجا که هر دو پروسه به خواب ابدی رفته‌اند، ما وارد بن‌بست شده‌ایم. بنابراین نتیجه این راه حل رضایت‌بخش نیست.
سمافور
سمافورها مشکل سیگنال wakeup گم‌شده را حل می‌کنند. در راه حل زیر ما از دو سمافور به نام‌های fillCount و emptyCount برای حل این مسئله استفاده می‌کنیم. fillCount تعداد اقلام موجود در بافر هستند که آماده خواندن هستند و emptyCount تعداد فضاهای خالی موجود در بافر است که می‌توان یک قلم داده را در هر فضا قرار داد. fillCount با قرار دادن یک قلم داده جدید در بافر افزایش می‌یابد. emptyCount هم با بداشتن یک قلم داده از بافر کاهش می‌یابد. اگر emptyCount صفر باشد و تولیدکننده بخواهد آن را کاهش دهد، تولید کننده به خواب خواهد رفت. دفعه بعد که یک قلم داده از بافر برداشته و مصرف شد، emptyCount یک واحد افزایش می‌یابد و تولیدکننده از خواب بیدار می‌شود. مصرف‌کننده هم به شکل مشابهی رفتار می‌کند.
semaphore fillCount = 0; // items produced
semaphore emptyCount = BUFFER_SIZE; // remaining space
procedure producer() {
while (true) {
item = produceItem();
down(emptyCount);
putItemIntoBuffer(item);
up(fillCount);
procedure consumer() {
while (true) {
down(fillCount);
item = removeItemFromBuffer();
up(emptyCount);
consumeItem(item);
وقتی که تنها یک مصرف کننده و یک تولید کننده وجود دارد، راه حل بالا به خوبی کار می‌کند. برای حالتی که چند تولیدکننده قسمت یکسانی از حافظه را به عنوان بافر به اشتراک می‌گذارند، یا چند مصرف‌کننده فضای یکسانی از حافظه را به اشتراک می‌گذارند، این راه حل یک وضعیت رقابتی جدی دارد که باعث می‌شود دو یا چند فرایند سعی کنند از یک قسمت بافر، داده‌ای را به شکل همزمان بخوانند یا بنویسند که ما را به سمت بن‌بست می‌کشاند. برای نشان دادن این مسئله، به پیاده‌سازی روال putItemIntoBuffer() دقت کنید. این روال دربرگیرنده دو عمل است. یکی تعیین می‌کند که اسلات خالی بعدی کدام است و دیگری داده‌ای را در آن اسلات می‌نویسد. اگر این روال به شکل همروند توسط چند تولیدکننده اجرا شود، ممکن است سناریو زیر اتفاق بیفتند:
دو تولیدکننده متغیر emptyCount را کاهش می‌دهند.
یکی از تولیدکننده‌ها اسلات خالی بعدی در بافر را تعیین می‌کند.
تولید کننده بعدی هم اسلات خالی بافر را تعیین می‌کند و همان نتیجه‌ای را می‌گیرد که تولید کننده قبلی گرفته است.
هر دو تولید کننده سعی می‌کنند در یک اسلات مشابه از بافر داده‌ای را قرار دهند.
برای حل این مشکل ما باید مطمئن شویم که تنها یک تولیدکننده می‌تواند در هر بار روال putItemIntoBuffer() را اجرا کند. اگر تولیدکننده‌ای مشغول اجرای این روال بود، تولیدکننده‌های دیگر باید منتظر بمانند تا تولیدکننده اول از این روال خارج شود. به عبارتی دیگر ما باید راهی پیدا کنیم که ناحیه بحرانی را با انحصار متقابل اجرا کنیم. برای انجام دادن این کار، ما از یک سمافور دودویی بنام mutex استفاده می‌کنیم. از آنجا که مقدار یک سمافور باینری تنها می‌تواند صفر یا ۱ باشد، تنها یک پروسه می‌تواند در هر بار down(mutex) یا up(mutex) را اجرا کند. راه حل مسئله بالا برای چند تولیدکننده و چند مصرف‌کننده در زیر ارائه شده است:
semaphore mutex = 1;
semaphore fillCount = 0;
semaphore emptyCount = BUFFER_SIZE;
procedure producer() {
while (true) {
item = produceItem();
down(emptyCount);
down(mutex);
putItemIntoBuffer(item);
up(mutex);
up(fillCount);
procedure consumer() {
while (true) {
down(fillCount);
down(mutex);
item = removeItemFromBuffer();
up(mutex);
up(emptyCount);
consumeItem(item);
ترتیبی که سمافورهای مختلف افزایش یا کاهش یافته‌اند بسیار مهم است. اگر ترتیب افزایش یا کاهش سمافورها تغییر کند، وارد بن‌بست خواهیم شد.
مثال‌ها
مثال از سی++

0; ))
product = products.top();
products.pop();
print(stringstream()
0)
consume(id);
this_thread::sleep_for(chrono::milliseconds(consumer_delay_to_consume));
print(stringstream()
= NO_ITEMS;
public void run() {
while (!theEnd()) {
synchronized (items) {
while (items.isEmpty() -- (!theEnd())) {
try {
items.wait(10);
catch (InterruptedException e) {
Thread.interrupted();
consume();