Comment implémenter le blocage de lecture à l'aide de threads POSIX

Question

Je souhaite implémenter un scénario producteur / consommateur qui obéit à des interfaces approximativement:

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

Dans ce cas, feed et run seront exécutés sur des threads distincts et read doit être une lecture bloquante (comme recv et fread ). De toute évidence, il me faudra une sorte d’exclusion mutuelle sur mon dossier, ainsi qu’un système de notification pour informer read et réessayer.

J'entends parler de variables de condition , mais toute mon expérience de multithreading réside dans Windows et j'ai du mal à comprendre pourquoi.

Merci pour toute aide!

(Oui, je sais qu'il est inefficace de renvoyer des vecteurs. N'entrons pas dans cela.)

Answer 1

Ce code n'est pas prêt pour la production. Aucune vérification d'erreur n'est effectuée sur les résultats des appels de bibliothèque.

J'ai emballé le verrou / déverrouillage du mutex dans LockThread afin qu'il soit protégé contre les exceptions. Mais c'est à peu près tout.

De plus, si j’agissais sérieusement, j’envelopperais le mutex et conditionnerait les variables à l’intérieur des objets afin qu’elles puissent être utilisées de manière abusive dans d’autres méthodes de Consumer. Mais tant que vous noterez que le verrou doit être acquis avant d’utiliser la variable de condition (de quelque manière que ce soit), cette situation simple peut rester telle quelle.

Désirez-vous vérifier la bibliothèque de threads boost?

#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

J'ai tendance à utiliser ce que j'appelle une "file d'attente synchronisée". J'emballe la file d'attente normale et utilise une classe Semaphore pour le verrouillage et la création d'un bloc de lecture, comme vous le souhaitez:

#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

Vous pouvez implémenter des sémaphores et des mutex avec les primitives appropriées dans votre implémentation de threading.

REMARQUE: cette implémentation est un exemple pour des éléments uniques dans une file d'attente, mais vous pouvez facilement l'envelopper avec une fonction qui tamponne les résultats jusqu'à ce que N ait été fourni. quelque chose comme ceci s'il s'agit d'une file d'attente de caractères:

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

Je vais jeter un semi-pseudo-code. Voici mes commentaires:

1) Très gros grains de verrouillage ici. Si vous avez besoin d'un accès plus rapide, vous voudrez repenser vos structures de données. La STL n’est pas threadsafe.

2) Le verrouillage sera bloqué jusqu'à ce que le mutex le laisse passer. La structure du mutex consiste à laisser passer un fil à la fois avec le mécanisme de verrouillage / déverrouillage. Pas besoin de sondage ou d'une sorte de structure d'exception.

3) C’est une jolie coupe syntaxiquement hacky du problème. Je ne suis pas précis avec la syntaxe API ni C ++, mais je crois que cela donne une solution sémantiquement correcte.

4) Modifié en réponse au commentaire.

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

Juste pour le plaisir, voici une implémentation rapide et sale utilisant Boost. Il utilise des pthreads sous le capot sur les plates-formes qui le prennent en charge et, sous Windows, utilise les opérations Windows.

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

Pour encore plus de plaisir, voici ma version finale. STL-ized sans raison valable. : -)

#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

Les files d'attente asynchrones Glib permettent le verrouillage et la veille lors de la lecture d'une file d'attente vide que vous recherchez. Voir http://library.gnome.org/devel/glib /2.20/glib-Asynchronous-Queues.html Vous pouvez les combiner avec des gthreads ou des pools de gthread.