Как реализовать блокировку чтения с помощью потоков POSIX
https://stackoverflow.com/questions/206857
-
03-07-2019 - |
italiano
english
français
española
中国
日本の
العربية
Deutsch
한국어
Português
RussianВопрос
Я хотел бы реализовать сценарий производителя / потребителя, который подчиняется интерфейсам, которые примерно:
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).Очевидно, что мне понадобится какое-то взаимное исключение в моем deque, и мне понадобится какая-то система уведомлений для информирования read попробовать еще раз.
Я слышу переменные условия это правильный путь, но весь мой опыт многопоточности связан с Windows, и мне трудно разобраться в них.
Спасибо за любую помощь!
(Да, я знаю, что возвращать векторы неэффективно.Давайте не будем вдаваться в подробности.)
Решение
Этот код не готов к производству.Проверка ошибок по результатам каких-либо вызовов библиотеки не выполняется.
Я обернул блокировку / разблокировку мьютекса в LockThread, чтобы он был безопасен для исключений.Но в этом-то все и дело.
Кроме того, если бы я занимался этим серьезно, я бы обернул переменные mutex и condition внутри объектов, чтобы ими можно было злоупотреблять в других методах Consumer.Но до тех пор, пока вы принимаете во внимание, что блокировка должна быть получена до того, как вы используете переменную условия (любым способом), тогда эта простая ситуация может оставаться как есть.
Из интереса вы проверили библиотеку boost threading library?
#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;
}
Другие советы
Я обычно использую то, что я называю "Синхронизированной очередью".Я оборачиваю обычную очередь и использую класс Semaphore как для блокировки, так и для создания блока чтения так, как вы хотите:
#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;
}
Я сброшу какой-нибудь полу-псевдокод.Вот мои комментарии:
1) Здесь очень крупные зерна запирания.Если вам нужен более быстрый доступ, вам захочется переосмыслить свои структуры данных.STL не является потокобезопасным.
2)Lock будет блокировать до тех пор, пока мьютекс не пропустит его.Структура мьютекса заключается в том, что он пропускает через себя 1 поток одновременно с помощью механизма блокировки / разблокировки.Нет необходимости в опросах или в какой-то структуре в виде исключений.
3) Это довольно синтаксически неаккуратный подход к проблеме.Я не уточняю ни 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)
}
Просто для развлечения, вот быстрая и грязная реализация с использованием Boost.Он использует pthreads под капотом на платформах, которые его поддерживают, а в Windows использует операции 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();
}
Для еще большего удовольствия, вот моя окончательная версия.Заблокирован без уважительной причины.:-)
#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);
}
};
Асинхронные очереди Glib обеспечивают блокировку и переход в спящий режим при чтении пустой очереди, которую вы ищете.Видишь http://library.gnome.org/devel/glib/2.20/glib-Asynchronous-Queues.html Вы можете объединить их с gthreads или пулами gthread.
