लागू करने के लिए कैसे अवरुद्ध POSIX धागे

Question

का उपयोग करके पढ़ा मैं एक निर्माता/उपभोक्ता परिदृश्य है कि इंटरफेस है कि मोटे तौर पर कर रहे हैं का अनुसरण करता है लागू करने के लिए करना चाहते हैं:

class Consumer { 
private: 
    vector<char> read(size_t n) { 
     // If the internal buffer has `n` elements, then dequeue them 
     // Otherwise wait for more data and try again 
    } 
public: 
    void run() { 
     read(10); 
     read(4839); 
     // etc 
    } 
    void feed(const vector<char> &more) { 
     // Safely queue the data 
     // Notify `read` that there is now more data 
    } 
}; 

इस मामले में, feed और run अलग धागे और read पर चलेगा चाहिए अवरुद्ध हो जाना (जैसे recv और fread)। जाहिर है, मुझे अपने डेक पर किसी प्रकार का पारस्परिक बहिष्कार की आवश्यकता होगी, और मुझे फिर से प्रयास करने के लिए read को सूचित करने के लिए किसी प्रकार की अधिसूचना प्रणाली की आवश्यकता होगी।

मुझे सुना है हालत चर जाने का रास्ता है, लेकिन मेरे सभी मल्टीथ्रेडिंग अनुभव विंडोज के साथ हैं और मेरे चारों ओर मेरे सिर को लपेटने में कठिनाई हो रही है।

किसी भी मदद के लिए धन्यवाद!

(हाँ, मैं जानता हूँ कि यह वैक्टर वापस जाने के लिए अक्षम है। चलो कि में मिलता है।)

Answer 1

यह कोड उत्पादन तैयार नहीं है। किसी लाइब्रेरी कॉल के परिणामों पर कोई त्रुटि जांच नहीं की जाती है।

मैंने लॉक थ्रेड में म्यूटेक्स के लॉक/अनलॉक को लपेट लिया है, इसलिए यह अपवाद सुरक्षित है। इसके बारे में बस इतना ही।

इसके अतिरिक्त यदि मैं इसे गंभीरता से कर रहा था तो मैं वस्तुओं के अंदर म्यूटेक्स और हालत चर को लपेटूंगा ताकि उपभोक्ता के अन्य तरीकों के अंदर उनका दुरुपयोग हो सके। लेकिन जब तक आप ध्यान दें कि कंडीशन वैरिएबल (किसी भी तरह से) का उपयोग करने से पहले लॉक अधिग्रहित किया जाना चाहिए, तो यह सरल स्थिति खड़ी हो सकती है।

ब्याज से आपने बूस्ट थ्रेडिंग लाइब्रेरी की जांच की है?

#include <iostream> 
#include <vector> 
#include <pthread.h> 

class LockThread 
{ 
    public: 
    LockThread(pthread_mutex_t& m) 
     :mutex(m) 
    { 
     pthread_mutex_lock(&mutex); 
    } 
    ~LockThread() 
    { 
     pthread_mutex_unlock(&mutex); 
    } 
    private: 
     pthread_mutex_t& mutex; 
}; 
class Consumer 
{ 
    pthread_mutex_t  lock; 
    pthread_cond_t  cond; 
    std::vector<char> unreadData; 
    public: 
    Consumer() 
    { 
     pthread_mutex_init(&lock,NULL); 
     pthread_cond_init(&cond,NULL); 
    } 
    ~Consumer() 
    { 
     pthread_cond_destroy(&cond); 
     pthread_mutex_destroy(&lock); 
    } 

    private: 
     std::vector<char> read(size_t n) 
     { 
      LockThread locker(lock); 
      while (unreadData.size() < n) 
      { 
       // Must wait until we have n char. 
       // This is a while loop because feed may not put enough in. 

       // pthread_cond() releases the lock. 
       // Thread will not be allowed to continue until 
       // signal is called and this thread reacquires the lock. 

       pthread_cond_wait(&cond,&lock); 

       // Once released from the condition you will have re-aquired the lock. 
       // Thus feed() must have exited and released the lock first. 
      } 

      /* 
      * Not sure if this is exactly what you wanted. 
      * But the data is copied out of the thread safe buffer 
      * into something that can be returned. 
      */ 
      std::vector<char> result(n); // init result with size n 
      std::copy(&unreadData[0], 
         &unreadData[n], 
         &result[0]); 

      unreadData.erase(unreadData.begin(), 
          unreadData.begin() + n); 
      return (result); 
     } 
public: 
    void run() 
    { 
     read(10); 
     read(4839); 
     // etc 
    } 
    void feed(const std::vector<char> &more) 
    { 
     LockThread locker(lock); 

     // Once we acquire the lock we can safely modify the buffer. 
     std::copy(more.begin(),more.end(),std::back_inserter(unreadData)); 

     // Only signal the thread if you have the lock 
     // Otherwise race conditions happen. 
     pthread_cond_signal(&cond); 

     // destructor releases the lock and thus allows read thread to continue. 
    } 
}; 


int main() 
{ 
    Consumer c; 
} 

Answer 2

मैं कुछ अर्द्ध छद्म कोड नीचे फेंक देंगे। यहां मेरी टिप्पणियां दी गई हैं:

1) यहां लॉक करने के बहुत बड़े अनाज। यदि आपको तेज़ी से पहुंच की आवश्यकता है, तो आप अपने डेटा संरचनाओं पर पुनर्विचार करना चाहेंगे। एसटीएल थ्रेडसेफ नहीं है।

2) लॉक तब तक अवरुद्ध होगा जब तक कि म्यूटेक्स इसे अनुमति न दे। म्यूटेक्स संरचना यह है कि यह लॉक/अनलॉक तंत्र के साथ एक समय में 1 धागा देता है। मतदान के लिए या किसी प्रकार की अपवाद-एस्क संरचना के लिए कोई ज़रूरत नहीं है।

3) यह समस्या पर एक सुंदर वाक्य रचनात्मक रूप से हैकी कट है। मैं एपीआई और सी ++ वाक्यविन्यास के साथ सटीक नहीं हूं, लेकिन मेरा मानना ​​है कि यह एक अर्थात् सही समाधान देता है।

4) टिप्पणी के जवाब में संपादित किया गया।

class piper 
{ 
pthread_mutex queuemutex; 
pthread_mutex readymutex; 
bool isReady; //init to false by constructor 

//whatever else 
}; 

piper::read() 
{//whatever 
pthread_mutex_lock(&queuemutex) 
if(myqueue.size() >= n) 
{ 
    return_queue_vector.push_back(/* you know what to do here */) 

    pthread_mutex_lock(&readymutex) 
    isReady = false; 
    pthread_mutex_unlock(&readymutex) 
} 
pthread_mutex_unlock(&queuemutex) 
} 

piper::push_em_in() 
{ 
//more whatever 
pthread_mutex_lock(&queuemutex) 
//push push push 
if(myqueue.size() >= n) 
{ 
    pthread_mutex_lock(&readymutex) 
    isReady = true; 
    pthread_mutex_unlock(&readymutex) 
} 
pthread_mutex_unlock(&queuemutex) 
} 

Answer 3

मैं जो "सिंक्रनाइज़ेड कतार" कहता हूं उसका उपयोग करता हूं। मैं सामान्य कतार लपेट और दोनों लॉकिंग के लिए एक सेमाफोर वर्ग का उपयोग करें और ब्लॉक पढ़ा जिससे आप इच्छा बस के रूप में:

#ifndef SYNCQUEUE_20061005_H_ 
#define SYNCQUEUE_20061005_H_ 

#include <queue> 
#include "Semaphore.h" 

// similar, but slightly simpler interface to std::queue 
// this queue implementation will serialize pushes and pops 
// and block on a pop while empty (as apposed to throwing an exception) 
// it also locks as neccessary on insertion and removal to avoid race 
// conditions 

template <class T, class C = std::deque<T> > class SyncQueue { 
protected: 
    std::queue<T, C> m_Queue; 
    Semaphore   m_Semaphore; 
    Mutex    m_Mutex; 

public: 
    typedef typename std::queue<T, C>::value_type value_type; 
    typedef typename std::queue<T, C>::size_type size_type; 

    explicit SyncQueue(const C& a = C()) : m_Queue(a), m_Semaphore(0) {} 

    bool empty() const    { return m_Queue.empty(); } 
    size_type size() const   { return m_Queue.size(); } 

    void push(const value_type& x); 
    value_type pop(); 
}; 

template <class T, class C> 
void SyncQueue<T, C>::push(const SyncQueue<T, C>::value_type &x) { 
    // atomically push item 
    m_Mutex.lock(); 
    m_Queue.push(x); 
    m_Mutex.unlock(); 

    // let blocking semaphore know another item has arrived 
    m_Semaphore.v(); 
} 

template <class T, class C> 
typename SyncQueue<T, C>::value_type SyncQueue<T, C>::pop() { 
    // block until we have at least one item 
    m_Semaphore.p(); 

    // atomically read and pop front item 
    m_Mutex.lock(); 
    value_type ret = m_Queue.front(); 
    m_Queue.pop(); 
    m_Mutex.unlock(); 

    return ret; 
} 

#endif 

आप सेमाफोर और अपने सूत्रण कार्यान्वयन में उचित पुरातन साथ mutexes लागू कर सकते हैं।

नोट: यह कार्यान्वयन कतार में एकल तत्वों के लिए एक उदाहरण है, लेकिन आप इसे आसानी से उस फ़ंक्शन के साथ लपेट सकते हैं जो एन प्रदान किए जाने तक परिणाम बफर करता है।ऐसा कुछ अगर यह वर्णों की कतार है:

std::vector<char> func(int size) { 
    std::vector<char> result; 
    while(result.size() != size) { 
     result.push_back(my_sync_queue.pop()); 
    } 
    return result; 
} 

Answer 4

बस मस्ती के लिए, यहां बूस्ट का उपयोग करके त्वरित और गंदा कार्यान्वयन है। यह प्लेटफार्मों पर हुड के नीचे pthreads का उपयोग करता है जो इसका समर्थन करता है, और विंडोज़ पर विंडोज़ ऑपरेशंस का उपयोग करता है।

boost::mutex access; 
boost::condition cond; 

// consumer 
data read() 
{ 
    boost::mutex::scoped_lock lock(access); 
    // this blocks until the data is ready 
    cond.wait(lock); 

    // queue is ready 
    return data_from_queue(); 
} 

// producer 
void push(data) 
{ 
    boost::mutex::scoped_lock lock(access); 
    // add data to queue 

    if (queue_has_enough_data()) 
    cond.notify_one(); 
} 

Answer 5

और भी मजेदार के लिए, मेरा अंतिम संस्करण यहां है। किसी भी अच्छे कारण के लिए एसटीएल-ized। :-)

#include <algorithm> 
#include <deque> 
#include <pthread.h> 

template<typename T> 
class MultithreadedReader { 
    std::deque<T> buffer; 
    pthread_mutex_t moreDataMutex; 
    pthread_cond_t moreDataCond; 

protected: 
    template<typename OutputIterator> 
    void read(size_t count, OutputIterator result) { 
     pthread_mutex_lock(&moreDataMutex); 

     while (buffer.size() < count) { 
      pthread_cond_wait(&moreDataCond, &moreDataMutex); 
     } 
     std::copy(buffer.begin(), buffer.begin() + count, result); 
     buffer.erase(buffer.begin(), buffer.begin() + count); 

     pthread_mutex_unlock(&moreDataMutex); 
    } 

public: 
    MultithreadedReader() { 
     pthread_mutex_init(&moreDataMutex, 0); 
     pthread_cond_init(&moreDataCond, 0); 
    } 

    ~MultithreadedReader() { 
     pthread_cond_destroy(&moreDataCond); 
     pthread_mutex_destroy(&moreDataMutex); 
    } 

    template<typename InputIterator> 
    void feed(InputIterator first, InputIterator last) { 
     pthread_mutex_lock(&moreDataMutex); 

     buffer.insert(buffer.end(), first, last); 
     pthread_cond_signal(&moreDataCond); 

     pthread_mutex_unlock(&moreDataMutex); 
    } 
}; 

Answer 6

ग्लिब असिंक्रोनस क्विज़ लॉकिंग और एक खाली कतार पढ़ने पर नींद प्रदान करते हैं जिसे आप ढूंढ रहे हैं। http://library.gnome.org/devel/glib/2.20/glib-Asynchronous-Queues.html देखें आप उन्हें gthreads या gthread पूल के साथ जोड़ सकते हैं।