निर्माता-उपभोक्ता समस्या

कम्प्यूटिंग में, निर्माता-उपभोक्ता समस्या (जिसे बाउंडेड-बफर समस्या के रूप में भी जाना जाता है) 1965 से एडजर डब्ल्यू. डिज्कस्ट्रा द्वारा वर्णित समस्याओं का वर्ग है।

डीजक्स्ट्रा ने निर्माता-उपभोक्ता समस्या का समाधान पाया क्योंकि उन्होंने इलेक्ट्रोलॉजिका X1 और X8 कंप्यूटरों के लिए एक सलाहकार के रूप में काम किया: निर्माता-उपभोक्ता का पहला उपयोग आंशिक रूप से सॉफ्टवेयर, आंशिक रूप से हार्डवेयर था: स्टोर और परिधीय के बीच सूचना परिवहन की देखभाल करने वाला घटक 'एक चैनल' कहा जाता था ... सिंक्रनाइज़ेशन को दो काउंटिंग सेमाफोर द्वारा नियंत्रित किया जाता था जिसे अब हम निर्माता/उपभोक्ता व्यवस्था के रूप में जानते हैं: एक सेमाफोर लाइन की लंबाई का संकेत देता है, CPU द्वारा (V में) बढ़ाया और घटाया गया था (एक P में) चैनल द्वारा, अन्य एक, अनजाने पूर्णताओं की संख्या की गणना करते हुए, चैनल द्वारा बढ़ाया गया था और CPU द्वारा घटाया गया था। [दूसरा सेमाफोर धनात्मक होने के कारण संबंधित इंटरप्ट फ्लैग को उठाएगा।]

दिज्क्स्ट्रा ने असीमित बफर की स्थिति के बारे में लिखा: हम दो प्रक्रियाओं पर विचार करते हैं, जिन्हें क्रमशः 'निर्माता' और 'उपभोक्ता' कहा जाता है। निर्माता एक चक्रीय प्रक्रिया है और हर बार जब यह अपने चक्र के माध्यम से जाता है है तो यह सूचना का एक निश्चित भाग उत्पन्न करता है, जिसे उपभोक्ता द्वारा संसाधित किया जाना है। उपभोक्ता भी चक्रीय प्रक्रिया है और हर बार जब वह अपने चक्र के माध्यम से जाता है, तो वह सूचना के अगले हिस्से को संसाधित कर सकता है, जैसा कि निर्माता द्वारा निर्मित किया गया है ... हम मानते हैं कि इस उद्देश्य के लिए दो प्रक्रियाओं को एक बफर के माध्यम से असीमित क्षमता के साथ जोड़ा जाना चाहिए।

उन्होंने सीमित बफर की स्थिति के बारे में लिखा: हमने निर्माता और उपभोक्ता को एक बफर के माध्यम से असीमित क्षमता के साथ अध्ययन किया है ... जो बफर के माध्यम से असीम क्षमता के साथ जुड़ा हुआ है . संबंध सममित हो जाता है, अगर दोनों परिमित आकार के बफर के माध्यम से जोड़े जाते हैं, N भाग कहते हैं।"

और कई निर्माता-उपभोक्ता की स्थिति के बारे में: हम कई निर्माता/उपभोक्ता जोड़े पर विचार करते हैं, जहां जोड़ी को एक सूचना धारा के माध्यम से जोड़ा जाता है जिसमें n_i भाग होते हैं। हम मानते हैं ... परिमित बफ़र जिसमें सभी धाराओं के सभी भाग सम्मिलित होने चाहिए, जिसमें 'टॉट' भाग की क्षमता होनी चाहिए।

प्रति ब्रिन्च हैनसेन और निकोलस विर्थ ने जल्द ही सेमाफोर की समस्या को देखा: मैं सेमाफोर के संबंध में एक ही निष्कर्ष पर पहुंचा हूं, अर्थात् वे उच्च स्तरीय भाषाओं के लिए उपयुक्त नहीं हैं। इसके अतिरिक्त, प्राकृतिक तुल्यकालन घटनाएँ संदेश का आदान-प्रदान हैं।

दिज्क्स्ट्रा का सीमित बफर समाधान

प्रारंभिक सेमाफोर सीमित बफर समाधान ALGOL शैली में लिखा गया था। बफर N भागों या तत्वों को स्टोर कर सकता है। कतारबद्ध भागों की संख्या सेमाफोर (प्रोग्रामिंग) बफर में भरे हुए स्थानों की गणना करता है, खाली पदों की संख्या सेमाफोर बफर में खाली स्थानों की गणना करता है और सेमाफोर बफर हेरफेर बफर पुट और ऑपरेशन प्राप्त करने के लिए म्युटेक्स के रूप में काम करता है। यदि बफ़र भरा हुआ है, यानी खाली स्थिति की संख्या शून्य है, तो निर्माता थ्रेड P (खाली स्थिति की संख्या) ऑपरेशन में प्रतीक्षा करेगा। यदि बफ़र खाली है, यानी कतारबद्ध भागों की संख्या शून्य है, तो उपभोक्ता धागा P (कतारबद्ध भागों की संख्या) ऑपरेशन में प्रतीक्षा करेगा। V() ऑपरेशन सेमाफोर जारी करते हैं। साइड इफेक्ट के रूप में, एक थ्रेड प्रतीक्षा कतार से तैयार कतार में जा सकता है। P() ऑपरेशन सेमाफोर मान को घटाकर शून्य कर देता है। V() ऑपरेशन सेमाफोर मान को बढ़ाता है।

begin integer number of queueing portions, number of empty positions,
      buffer manipulation;
      number of queueing portions:= 0;
      number of empty positions:= N;
      buffer manipulation:= 1;
      parbegin
      producer: begin
              again 1: produce next portion;
                       P(number of empty positions);
                       P(buffer manipulation);
                       add portion to buffer;
                       V(buffer manipulation);
                       V(number of queueing portions); goto again 1 end;
      consumer: begin
              again 2: P(number of queueing portions);
                       P(buffer manipulation);
                       take portion from buffer;
                       V(buffer manipulation) ;
                       V(number of empty positions);
                       process portion taken; goto again 2 end
      parend
end

C++ 20 के अनुसार, सेमाफोर भाषा का भाग हैं। डीजक्स्ट्रा के समाधान को आधुनिक C++ में आसानी से लिखा जा सकता है। परिवर्तनीय बफर_मैनिपुलेशन म्यूटेक्स है। एक थ्रेड में अधिग्रहण करने और दूसरे थ्रेड में रिलीज करने की सेमफोर विशेषता की आवश्यकता नहीं है। लॉक() और अनलॉक() जोड़ी के अतिरिक्त लॉक_गार्ड() कथन C++ RAII है। लॉक_गार्ड डिस्ट्रक्टर अपवाद की स्थिति में लॉक रिलीज सुनिश्चित करता है। यह समाधान कई उपभोक्ता थ्रेड और/या कई निर्माता थ्रेड को संभाल सकता है।

#include <thread>

#include <mutex>

#include <semaphore>

std::counting_semaphore<N> number_of_queueing_portions{0};

std::counting_semaphore<N> number_of_empty_positions{N};

std::mutex buffer_manipulation;

 void producer() {

  for (;;) {
    Portion portion = produce_next_portion();
    number_of_empty_positions.acquire();
    {
      std::lock_guard<std::mutex> g(buffer_manipulation);
      add_portion_to_buffer(portion);
    }
    number_of_queueing_portions.release();
  }
}

void consumer() {
  for (;;) {
    number_of_queueing_portions.acquire();
    Portion portion;
    {
      std::lock_guard<std::mutex> g(buffer_manipulation);
      portion = take_portion_from_buffer();
    }
    number_of_empty_positions.release();
    process_portion_taken(portion);
  }
}

int main() {
  std::thread t1(producer);
  std::thread t2(consumer);
  t1.join();
  t2.join();
}

मॉनिटर का प्रयोग

प्रति ब्रिंच हैनसेन ने मॉनिटर को परिभाषित किया: मैं एक साझा परिवर्ती और उस पर सार्थक संचालन के सेट को निरूपित करने के लिए मॉनिटर शब्द का उपयोग करूंगा। एक मॉनिटर का उद्देश्य निश्चित नीति के अनुसार अलग-अलग प्रक्रियाओं के बीच संसाधनों की समयबद्धता को नियंत्रित करना है। टोनी होरे ने मॉनिटर के लिए सैद्धांतिक नींव रखी थी।

bounded buffer: monitor
  begin buffer:array 0..N-1 of portion;
    head, tail: 0..N-1;
    count: 0..N;
    nonempty, nonfull: condition;
  procedure append(x: portion);
    begin if count = N then nonfull.wait;
      note 0 <= count < N;
      buffer[tail] := x;
      tail := tail (+) 1;
      count := count + 1;
      nonempty.signal
    end append;
  procedure remove(result x: portion) ;
    begin if count = 0 then nonempty.wait;
      note 0 < count <= N;
      x := buffer[head];
      head := head (+) 1;
      count := count - 1;
      nonfull.signal
    end remove;
  head := 0; tail := 0; count := 0;
end bounded buffer;

मॉनिटर एक वस्तु है जिसमें परिवर्ती होते हैं बफर,हेड,टेल और काउंट एक परिपत्र बफर का एहसास करने के लिए, स्थिति परिवर्ती जो तुल्यकालन के लिए अरिक्त और नॉनफुल होता है और विधियों को संलग्न और निर्दिष्ट बफर तक पहुंचने के लिए हटा देता है। मॉनिटर ऑपरेशन वेट सेमाफोर ऑपरेशन P, सिग्नल V या रिलीज अधिग्रहण के अनुरूप है। गोलाकार ऑपरेशन (+) को मोडुलो N लिया जाता है। प्रस्तुत पास्कल शैली सूडो कोड एक होयर मॉनिटर दिखाता है। एक मेसा (प्रोग्रामिंग भाषा) गिनती के अतिरिक्त मॉनिटर उपयोग करता है। एक प्रोग्रामिंग भाषा C++ संस्करण है:

  class Bounded_buffer {

  Portion buffer[N];    // 0..N-1
  unsigned head, tail;  // 0..N-1
  unsigned count;       // 0..N
  std::condition_variable nonempty, nonfull;
  std::mutex mtx;
public:
  void append(Portion x) {
    std::unique_lock<std::mutex> lck(mtx);
    nonfull.wait(lck, [&]{ return!(N == count); });
    assert(0 <= count && count < N);
    buffer[tail++] = x;
    tail %= N;
    ++count;
    nonempty.notify_one();
  }
  Portion remove() {
    std::unique_lock<std::mutex> lck(mtx);
    nonempty.wait(lck, [&]{ return!(0 == count); });
    assert(0 < count && count <= N);
    Portion x = buffer[head++];
    head %= N; 
    --count;
    nonfull.notify_one();
    return x;
  }
  Bounded_buffer() {
    head = 0; tail = 0; count = 0;
  }
};

C++ संस्करण को तकनीकी कारणों के अतिरिक्त म्यूटेक्स की आवश्यकता है। यह बफ़र जोड़ने और हटाने के संचालन के लिए पूर्व शर्त लागू करने के लिए दृढ़ता का उपयोग करता है।

चैनलों का उपयोग करना

इलेक्ट्रोलॉजिका कंप्यूटरों में सबसे पहले निर्माता-उपभोक्ता समाधान ने 'चैनल' का उपयोग किया। होरे परिभाषित चैनल (प्रोग्रामिंग): स्रोत और गंतव्य के स्पष्ट नामकरण का एक विकल्प एक पोर्ट का नाम देना होगा जिसके माध्यम से संचार होना है। पोर्ट नाम प्रक्रियाओं के लिए स्थानीय होंगे, और जिस तरीके से पोर्ट के जोड़े को चैनलों से जोड़ा जाना है, उसे समानांतर कमांड के प्रमुख में घोषित किया जा सकता है। ब्रिन्च हैनसेन ने प्रोग्रामिंग भाषाओं जॉयस (प्रोग्रामिंग भाषा) संचार और सुपर पास्कल चैनल और संचार में चैनलों को कार्यान्वित किया। प्लान 9 ऑपरेटिंग सिस्टम प्रोग्रामिंग लैंग्वेज एलेफ (प्रोग्रामिंग लैंग्वेज), इन्फर्नो ऑपरेटिंग सिस्टम प्रोग्रामिंग लैंग्वेज लिम्बो (प्रोग्रामिंग भाषा) में चैनल हैं। निम्नलिखित C स्रोत कोड यूजर स्पेस से प्लान 9 पर संकलित है:

#include "u.h"
#include "libc.h"
#include "thread.h"

enum { STACK = 8192 };

void producer(void *v) {
  Channel *ch = v;
  for (uint i = 1; ; ++i) {
    sleep(400);
    print("p %d\n", i);
    sendul(ch, i);
  }
}
void consumer(void *v) {
  Channel *ch = v;
  for (;;) {
    uint p = recvul(ch);
    print("\t\tc %d\n", p);
    sleep(200 + nrand(600));
  }
}
void threadmain(int argc, char **argv) {
  int (*mk)(void (*fn)(void*), void *arg, uint stack);
  mk = threadcreate;
  Channel *ch = chancreate(sizeof(ulong), 1);
  mk(producer, ch, STACK);
  mk(consumer, ch, STACK);
  recvp(chancreate(sizeof(void*), 0));
  threadexitsall(0);
}

फलन का प्रवेश बिंदुथ्रेडमेनकार्य कर रहा है, फ़ंक्शन कॉल ch = chancreate(sizeof(ulong), 1) चैनल बनाता है, फ़ंक्शन कॉल करता है sendul(ch, i) चैनल और फ़ंक्शन कॉल में मान भेजता है p = recvul(ch) चैनल से मूल्य प्राप्त करता है। प्रोग्रामिंग लैंग्वेज गो (प्रोग्रामिंग लैंग्वेज) में चैनल भी हैं। एक गो उदाहरण:

package main

import (
	"fmt"
	"math/rand"
	"time"
)

var sendMsg = 0

func produceMessage() int {
	time.Sleep(400 * time.Millisecond)
	sendMsg++
	fmt.Printf("sendMsg = %v\n", sendMsg)
	return sendMsg
}
func consumeMessage(recvMsg int) {
	fmt.Printf("\t\trecvMsg = %v\n", recvMsg)
	time.Sleep(time.Duration(200+rand.Intn(600)) * time.Millisecond)
}
func main() {
	ch := make(chan int, 3)
	go func() {
		for {
			ch <- produceMessage()
		}
	}()
	for recvMsg := range ch {
		consumeMessage(recvMsg)
	}
}

गो निर्माता-उपभोक्ता समाधान उपभोक्ता के लिए मुख्य गो रूटीन का उपयोग करता है और निर्माता के लिए एक नया, अनाम गो रूटीन बनाता है। दो गो रूटीन चैनल ch से जुड़े हुए हैं। यह चैनल तीन इंट वैल्यू तक कतारबद्ध कर सकता है। कथन ch:= make(chan int, 3) चैनल बनाता है, बयान ch <- produceMessage() चैनल और स्टेटमेंट में वैल्यू भेजता है recvMsg:= range ch चैनल से मूल्य प्राप्त करता है। स्मृति संसाधनों का आवंटन, प्रसंस्करण संसाधनों का आवंटन और संसाधनों का सिंक्रनाइज़ेशन स्वचालित रूप से प्रोग्रामिंग भाषा द्वारा किया जाता है।

सेमाफोर या मॉनिटर के बिना

लेस्ली लामपोर्ट ने एक निर्माता और एक उपभोक्ता के लिए बाध्य बफर निर्माता-उपभोक्ता समाधान प्रलेखित किया: हम मानते हैं कि b बफर अधिकतम b >= 1 संदेशों को पकड़ सकता है। हमारे समाधान में, हम k को b से अधिक निरंतर होने देते हैं, और चलो s और r पूर्णांक परिवर्ती हैं जो 0 और k-1 के बीच मान मानते हैं। हम मानते हैं कि शुरू में s=r और बफर खाली है। k को b का एक बहु होने के लिए चुनकर, बफर को एक सरणी B [0: b - 1] के रूप में लागू किया जा सकता है। निर्माता बस प्रत्येक नए संदेश को B[s mod b] में डालता है, और उपभोक्ता प्रत्येक संदेश को B[r mod b] से लेता है। एल्गोरिथ्म नीचे दिखाया गया है, अपरिमित k के लिए सामान्यीकृत किया गया है।

Producer:
  L:  if (s - r) mod k = b then goto L fi;
      put message in buffer;
      s := (s + 1) mod k;
      goto L;
Consumer:
  L:  if (s - r) mod k = 0 then goto L fi;
      take message from buffer;
      r := (r + 1) mod k;
      goto L;

लैमपोर्ट समाधान शेड्यूलर में प्रतीक्षा करने के अतिरिक्त थ्रेड में व्यस्त प्रतीक्षा का उपयोग करता है। यह समाधान असुविधाजनक समय पर शेड्यूलर थ्रेड स्विच के प्रभाव की उपेक्षा करता है। यदि पहले थ्रेड ने स्मृति से एक परिवर्ती मूल्य पढ़ा है, शेड्यूलर दूसरे थ्रेड पर स्विच करता है जो परिवर्ती मूल्य को बदलता है, और शेड्यूलर पहले थ्रेड में वापस स्विच करता है, तो पहला थ्रेड वेरिएबल के पुराने मूल्य न कि वर्तमान मूल्य का उपयोग करता है। परमाणु रीड-मोडिफाई-राइट इस समस्या को हल करते हैं। आधुनिक C++ ऑफर अटामिक मल्टी-थ्रेड प्रोग्रामिंग के लिए परमाणु परिवर्ती और संचालन प्रदान करता है। एक निर्माता और एक उपभोक्ता के लिए निम्नलिखित व्यस्त प्रतीक्षा C++11 समाधान एटम रीड-मोडिफाई-राइट ऑपरेशन fetch_add और fetch_sub का उपयोग करता है।

 enum {N = 4 };

Message buffer[N];
std::atomic<unsigned> count {0};
void producer() {
  unsigned tail {0};
  for (;;) {
    Message message = produceMessage();
    while (N == count) 
     ; // busy waiting
    buffer[tail++] = message;
    tail %= N;
    count.fetch_add(1, std::memory_order_relaxed);
  }
}
void consumer() {
  unsigned head {0};
  for (;;) {
    while (0 == count) 
     ; // busy waiting
    Message message = buffer[head++];
    head %= N;
    count.fetch_sub(1, std::memory_order_relaxed);
    consumeMessage(message);
  }
}
int main() {
  std::thread t1(producer);
  std::thread t2(consumer);
  t1.join();
  t2.join();
}

परिपत्र बफर सूचकांक परिवर्ती हेड और टेल थ्रेड-लोकल हैं और इसलिए मेमोरी स्थिरता के लिए प्रासंगिक नहीं हैं। परिवर्तीकाउन्ट निर्माता और उपभोक्ता थ्रेड की व्यस्त प्रतीक्षा को नियंत्रित करता है।

यह भी देखें

• परमाणु संचालन
• डिजाइन पैटर्न (कंप्यूटर विज्ञान)
• फीफो (कंप्यूटिंग और इलेक्ट्रॉनिक्स)
• पाइपलाइन (सॉफ्टवेयर)
• चैनल (प्रोग्रामिंग)
• जावा में कार्यान्वयन: जावा संदेश सेवा

अग्रिम पठन

• Mark Grand Patterns in Java, Volume 1, A Catalog of Reusable Design Patterns Illustrated with UML
• C/C++ Users Journal (Dr.Dobb's) January 2004, "A C++ Producer-Consumer Concurrency Template Library", by Ted Yuan, is a ready-to-use C++ template library. The small template library source code and examples can be found here
• Ioan Tinca, The Evolution of the Producer-Consumer Problem in Java