Как реализовать чтение блокирования с помощью потоков POSIX

Я брошу вниз некоторый semi-pseudo-code. Вот мои комментарии:

1) Очень большие мелкие частицы блокировки здесь. При необходимости в более быстром доступе Вы захотите заново продумать свои структуры данных. STL не ориентирован на многопотоковое исполнение.

2) Блокировка заблокируется, пока взаимное исключение не пропускает ее. Взаимоисключающая структура - то, что это позволяет 1 потоку через него за один раз с блокировать/разблокировать механизмом. Никакая потребность в опросе или в некоторой структуре исключения-esque.

3) Это довольно синтаксически hacky сокращение в проблеме. Я не точен с API, ни синтаксисом C++, но я полагаю, что он дает семантически правильное решение.

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

Еще для большего количества забавы вот моя окончательная версия. STL-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);
    }
};

Только для забавы, вот быстрая и грязная реализация с помощью Повышения. Это использует 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();  
}

Я склонен использовать то, что я называю "Очередью Syncronized". Я переношу нормальную очередь и использую Семафорный класс и для блокировки и для создания блока чтения, как Вы требуете:

#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

Можно реализовать семафоры и взаимные исключения с соответствующими примитивами в реализации поточной обработки.

Примечание: эта реализация является примером для единственных элементов в очереди, но Вы могли легко перенести это с функцией, которая буферизует результаты, пока N не были обеспечены. что-то вроде этого, если это - очередь символов:

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

Асинхронные очереди Glib обеспечивают блокировку и спящий режим при чтении пустой очереди, которую вы ищем. См. http://library.gnome.org/devel/glib/2.20/glib-Asynchronous-Queues.html Вы можете комбинировать их с gthreads или пулами gthread.