Wie Implementieren lesen Blockieren POSIX Threads

Question

würde Ich mag ein Producer / Consumer-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 separate Threads und read laufen sollte eine Blockierung (wie recv und fread) gelesen werden. Natürlich, ich werde eine Art gegenseitigen Ausschlusses auf meine deque brauchen, und ich werde eine Art von Benachrichtigungssystem müssen read informieren erneut zu versuchen.

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

Vielen Dank für jede Hilfe!

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

Answer 1

Dieser Code ist nicht die Produktion bereit. Keine Fehlerüberprüfung auf die Ergebnisse von Bibliotheksaufrufe durchgeführt wird.

Ich habe das Sperren / Entsperren des Mutex in LockThread gewickelt, damit es Ausnahme sicher ist. Aber das ist es.

Außerdem, wenn ich tat dies ernsthaft würde ich die Mutex und Bedingungsvariablen innerhalb Objekte wickeln, so können sie in andere Methoden der Verbrauchermissbraucht werden Bett. Aber solange man zur Kenntnis nehmen, dass das Schloss erworben werden muß, bevor Sie die Bedingung Variable (in irgendeiner Weise), dann diese einfache Situation stehen kann wie es ist.

Aus 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 neige dazu zu verwenden, was ich eine „Synchronisierte Queue“ nennen. Ich die normale Warteschlange wickeln und eine Semaphore-Klasse für beide Sperren und macht Leseblock verwenden, 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 dem entsprechenden Primitiven in Ihrer Einfädeln Implementierung implementieren.

Hinweis: Diese Implementierung ist ein Beispiel für einzelne Elemente in einer Warteschlange, aber man kann dies leicht mit einer Funktion wickelt die Ergebnisse puffert, bis N zur Verfügung gestellt wurde. so etwas wie diese, wenn es eine Warteschlange von Zeichen:

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

Answer 3

ich hinwerfen werde einige semi-Pseudo-Code. Hier sind meine Kommentare:

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

2) Sperre blockiert, bis der Mutex es durchlässt. Die Mutex Struktur ist, dass es läßt 1 Faden durch sie zu einer Zeit mit der Verriegelungs- / Entriegelungsmechanismus. Keine Notwendigkeit für den Abruf oder für eine Art von Ausnahme-artige Struktur.

3) Dies ist ein ziemlich syntaktisch Hacky Schnitt am Problem. Ich meine es genau nicht mit der API noch C ++ Syntax, aber ich glaube, es ist eine semantisch korrekte Lösung gibt.

4) in Reaktion Herausgegeben Stellung zu nehmen.

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 4

Just for fun, hier ist eine schnelle und schmutzige Implementierung mit boost. Es verwendet pthreads unter der Haube auf Plattformen, die es unterstützen, und an den Fenstern 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 letzte Version. STL-ized für keinen 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 Asynchronous Warteschlangen stellen die Verriegelung und schlafen auf eine leere Warteschlange zu lesen, 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.