Wie implementieren Blockierung lesen POSIX Threads

Question

Ich möchte einen Erzeuger/Verbraucher-Szenario implementieren, die Schnittstellen, die grob sind gehorcht:

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 
    } 
}; 

In diesem Fall feed und run auf separaten Threads ausgeführt werden und read sollte ein blockierender Lesevorgang sein (wie recv und fread). Offensichtlich brauche ich eine Art gegenseitigen Ausschluss auf meiner Deque, und ich werde eine Art von Benachrichtigungssystem benötigen, um read zu informieren, um es erneut zu versuchen.

Ich höre Zustand Variablen sind der Weg zu gehen, aber alle meine Multithreading-Erfahrung liegt bei Windows und habe eine harte Zeit, um meinen Kopf um sie zu wickeln.

Danke für jede Hilfe!

(Ja, ich weiß, dass es ineffizient ist Vektoren zurückzukehren. Lassen Sie uns nicht in das bekommen.)

Answer 1

Dieser Code ist nicht produktionsbereit. Es wird keine Fehlerüberprüfung der Ergebnisse von Bibliotheksaufrufen durchgeführt.

Ich habe das Lock/Unlock des Mutex in LockThread gewickelt, so dass es ausnahmslos sicher ist. Aber das war es schon.

Darüber hinaus würde ich die Mutex und Condition-Variablen in Objekte wickeln, so dass sie in anderen Methoden von Consumer missbraucht werden können. Aber solange Sie zur Kenntnis nehmen, dass die Sperre erworben werden muss, bevor Sie die Zustandsvariable (in irgendeiner Weise) verwenden, kann diese einfache Situation unverändert bestehen bleiben.

Von Interesse haben Sie die Boost-Threading-Bibliothek überprüft?

#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

Ich werde einige semi-Pseudo-Code werfen. Hier sind meine Kommentare:

1) Sehr große Körner der Verriegelung hier. Wenn Sie einen schnelleren Zugriff benötigen, sollten Sie Ihre Datenstrukturen überdenken. Die STL ist nicht threadsicher.

2) Die Sperre wird blockiert, bis der Mutex es durchlässt. Die Mutex-Struktur besteht darin, dass man sie mit dem Lock/Unlock-Mechanismus einmal durchlaufen kann. Keine Abfrage oder irgendeine ausnahmslose Struktur.

3) Dies ist eine ziemlich syntaktisch hacky Schnitt bei dem Problem. Ich bin nicht genau mit der API oder C++ - Syntax, aber ich glaube, es gibt eine semantisch korrekte Lösung.

4) Bearbeitet als Antwort auf den Kommentar.

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

Ich neige dazu zu verwenden, was ich eine "Syncronized Queue" nenne. Ich die normale Warteschlange wickeln und eine Semaphore-Klasse für beide Sperr verwenden und Block lesen zu machen wie Sie es wünschen:

#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 

Sie können Semaphore und Mutex mit den entsprechenden Primitiven in Ihrer Threading Implementierung implementieren.

ANMERKUNG: Diese Implementierung ist ein Beispiel für einzelne Elemente in einer Warteschlange, aber Sie könnten dies einfach mit einer Funktion umbrechen, die Ergebnisse puffert, bis N bereitgestellt wurde.etwas wie das, wenn es eine Reihe von Zeichen ist:

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

Nur zum Spaß, hier ist eine schnelle und schmutzige Umsetzung mit Boost. Es verwendet Pthreads unter der Haube auf Plattformen, die es unterstützen, und auf Windows verwendet Windows-Operationen.

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

Für noch mehr Spaß, hier ist meine endgültige Version. STL-iert ohne guten Grund. :-)

#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

Glib Asynchrone Warteschlangen bieten das Sperren und Schlaf beim Lesen einer leeren Warteschlange, die Sie suchen. Siehe http://library.gnome.org/devel/glib/2.20/glib-Asynchronous-Queues.html Sie können sie mit gthreads oder gthread-pools kombinieren.