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

مسئله تولیدکننده-مصرف‌کننده (به انگلیسی: Producer–consumer problem) یا مسئله بافر محدود (به انگلیسی: bounded-buffer problem) مسئله‌ای کلاسیک در همگام‌سازی چندپروسه‌ای است که این پروسه‌ها یک ناحیه بحرانی مشترک (به انگلیسی: Critical Section) دارند.

تشریح

در این مسئله، دو فرایند به نام‌های تولیدکننده و مصرف‌کننده وجود دارد که یک بافر با اندازه‌ای ثابت در بین آن‌ها به اشتراک گذاشته شده و آن‌ها از این بافر به عنوان صف استفاده می‌کنند. وظیفه تولیدکننده این است که یک قلم داده تولید کرده و در بافر قرار دهد. به شکل همزمان، مصرف‌کننده باید این قلم داده را از بافر بردارد و آن را به مصرف برساند. تولیدکننده و مصرف‌کننده تنها می‌توانند هر بار یک قلم داده را در بافر قرار دهند یا از بافر بردارند. مسئله اینجاست که ما باید مطمئن شویم که تولیدکننده سعی نمی‌کند در یک بافر پر شده، داده‌ای را ذخیره کند یا به‌طور مشابه، مصرف‌کننده سعی نکند داده‌ای را از یک بافر خالی بردارد. هرگاه که بافر پر بود و تولیدکننده خواست چیزی را در آن ذخیره کند، باید به خواب برود تا اینکه مصرف‌کننده یک قلم داده را از بافر بردارد و بعد از آن تولیدکننده را از خواب بیدار کند. به‌طور مشابه، اگر مصرف‌کننده خواست یک قلم داده را از یک بافر خالی بردارد، باید به خواب برود تا تولیدکنده یک قلم داده را در بافر قرار دهد و سپس مصرف‌کننده را بیدار کند. یک راه حل نامناسب برای این مسئله، می‌تواند به بن‌بست منجر شود که در آن هر دو پروس همزمان به خواب می‌روند و هیچکس نیست تا آن‌ها را از خواب بیدار کند که باید جلوی این اتفاق را گرفت. این مسئله را می‌توان به n تولیدکننده و m مصرف‌کننده تعمیم داد.

مثال از سی++

-pthread -lpthread -o pc*/#include <iostream>#include <sstream>#include <vector>#include <stack>#include <thread>#include <mutex>#include <atomic>#include <condition_variable>#include <chrono>using namespace std;// print function for "thread safe" printing using a stringstreamvoid print(ostream& s) { cout <<s.rdbuf(); cout.flush(); s.clear(); }//      Constants//const int num_producers = 5;const int num_consumers = 10;const int producer_delay_to_produce = 10;   // in millisecondsconst int consumer_delay_to_consume = 30;   // in millisecondsconst int consumer_max_wait_time = 200;     // in milliseconds - max time that a 
                                            // consumer can wait for a product to be produced.const int max_production = 10;              // When producers has produced this quantity 
                                            // they will stop to produceconst int max_products = 10;                // Maximum number of products that can be stored////      Variables//atomic<int> num_producers_working(0);       // When there's no producer working the consumers 
                                            // will stop, and the program will stop.stack<int> products;                        // The products stack, here we will store our productsmutex xmutex;                               // Our mutex, without this mutex our program will crycondition_variable is_not_full;             // to indicate that our stack is not full between 
                                            // the thread operationscondition_variable is_not_empty;            // to indicate that our stack is not empty between 
                                            // the thread operations////      Functions////      Produce function, producer_id will produce a productvoid produce(int producer_id){
        unique_lock<mutex> lock(xmutex);
        int product;

        is_not_full.wait(lock, [] { return products.size() != max_products; });
        product = products.size();
        products.push(product);

        print(stringstream() <<"Producer " <<producer_id <<" produced " <<product <<"\n");
        is_not_empty.notify_all();}//      Consume function, consumer_id will consume a productvoid consume(int consumer_id){
        unique_lock<mutex> lock(xmutex);
        int product;

        if(is_not_empty.wait_for(lock, chrono::milliseconds(consumer_max_wait_time),
                [] { return products.size()> 0; }))
        {
                product = products.top();
                products.pop();

                print(stringstream() <<"Consumer " <<consumer_id <<" consumed " <<product <<"\n");
                is_not_full.notify_all();
        }}//      Producer function, this is the body of a producer threadvoid producer(int id){
        ++num_producers_working;
        for(int i = 0; i <max_production; ++i)
        {
                produce(id);
                this_thread::sleep_for(chrono::milliseconds(producer_delay_to_produce));
        }

        print(stringstream() <<"Producer " <<id <<" has exited\n");
        --num_producers_working;}//      Consumer function, this is the body of a consumer threadvoid consumer(int id){
        // Wait until there is any producer working
        while(num_producers_working == 0) this_thread::yield();

        while(num_producers_working != 0 || products.size()> 0)
        {
                consume(id);
                this_thread::sleep_for(chrono::milliseconds(consumer_delay_to_consume));
        }

        print(stringstream() <<"Consumer " <<id <<" has exited\n");}////      Main//int main(){
        vector<thread> producers_and_consumers;

        // Create producers
        for(int i = 0; i <num_producers; ++i)
                producers_and_consumers.push_back(thread(producer, i));

        // Create consumers
        for(int i = 0; i <num_consumers; ++i)
                producers_and_consumers.push_back(thread(consumer, i));

        // Wait for consumers and producers to finish
        for(auto& t : producers_and_consumers)
                t.join();}

مثال از جاوا

import java.util.Stack;import java.util.concurrent.atomic.AtomicInteger;/** * 1 producer and 3 consumers producing/consuming 10 items */public class ProducerConsumer {

        Stack<Integer> items = new Stack<Integer>();
        final static int NO_ITEMS = 10;

        public static void main(String args[]) {
                ProducerConsumer pc = new ProducerConsumer();
                Thread t1 = new Thread(pc.new Producer());
                Consumer consumer  = pc.new Consumer();
                Thread t2 = new Thread(consumer);
                Thread t3 = new Thread(consumer);
                Thread t4 = new Thread(consumer);
                t1.start();
                try {
                        Thread.sleep(100);
                } catch (InterruptedException e1) {
                        e1.printStackTrace();
                }
                t2.start();
                t3.start();
                t4.start();
                try {
                        t2.join();
                        t3.join();
                        t4.join();
                } catch (InterruptedException e) {
                        e.printStackTrace();
                }
        }

        class Producer implements Runnable {

                public void produce(int i) {
                        System.out.println("Producing " + i);
                        items.push(new Integer(i));
                }

                public void run() {
                        int i = 0;
                        // produce 10 items
                        while (i++ <NO_ITEMS) {
                                synchronized (items) {
                                        produce(i);
                                        items.notifyAll();
                                }
                                try {
                                        // sleep for some time, 
                                        Thread.sleep(10);
                                } catch (InterruptedException e) {
                                }
                        }
                }
        }

        class Consumer implements Runnable {
                //consumed counter to allow the thread to stop
                AtomicInteger consumed = new AtomicInteger();

                public void consume() {
                        if (!items.isEmpty()) {
                                System.out.println("Consuming " + items.pop());
                                consumed.incrementAndGet();
                        }
                }

                private boolean theEnd() {
                        return consumed.get()>= NO_ITEMS;
                }

                public void run() {
                        while (!theEnd()) {
                                synchronized (items) {
                                        while (items.isEmpty() && (!theEnd())) {
                                                try {
                                                        items.wait(10);
                                                } catch (InterruptedException e) {
                                                        Thread.interrupted();
                                                }
                                        }
                                        consume();

                                }
                        }
                }
        }}