Быстрый производитель и медленный потребитель

У меня есть вариант использования, когда быстрый производитель вставляет данные в очередь, а медленный потребитель потребляет данные. Проблема, с которой я сталкиваюсь, - это постоянное увеличение размера очереди с течением времени. У меня есть реализация класса, в которой std :: queue защищена мьютексом и условной переменной для одновременных операций чтения и записи.

Как это можно адаптировать к случаю, когда производитель после достижения MAX_THRESHOLD до прекращения вставки данных в очередь, а потребитель израсходовал некоторое количество сигналов данных, чтобы производитель вставил данные в очередь.

Может ли кто-нибудь предоставить образец реализации?

Также без изменения реализации класса можно ли решить эту проблему, добавив еще один уровень синхронизации в производителе и потребителе?


multithreading pthreads concurrency producer-consumer
person KodeWarrior    schedule 10.03.2013    source источник
comment
См. Также programmers.stackexchange.com/q/244826/24257   -  person Pacerier    schedule 14.09.2015

Ответы (2)


arrow_upward
3
arrow_downward

Фрагмент кода «ограниченной очереди» с pthread:

#include <queue>
#include <pthread.h>

template <class T, size_t UpperLimit>
class BoundedQueue {
  std::queue<T> q_;
  pthread_mutex_t mtx_;
  pthread_cond_t cv_not_empry_;
  pthread_cond_t cv_not_full_;

  // lock/unlock helper
  struct auto_locker {
    auto_locker(pthread_mutex_t* pm) : pm_(pm)
      { pthread_mutex_lock(pm_); }
    ~auto_locker()
      { pthread_mutex_unlock(pm_);}
    pthread_mutex_t *pm_;
  };

public:
  BoundedQueue() { /* initialize member... */ }
  ~BoundedQueue() { /* uninitialize member...*/ }
  // for Producer
  void push(T x) {
    auto_locker lk(&mtx_);
    while (UpperLimit <= q_.size()) {
      pthread_cond_wait(&cv_not_full_, &mtx_);
    }
    q_.push(x);
    pthread_cond_broadcast(&cv_not_empry_);
    return ret;
  }
  // for Consumer
  T pop() {
    auto_locker lk(&mtx_);
    while (q_.empty()) {
      pthread_cond_wait(&cv_not_empry_, &mtx_);
    }
    T ret = q_.front();
    q_.pop();
    pthread_cond_broadcast(&cv_not_full_);
    return ret;
  }
}
person yohjp    schedule 11.03.2013

arrow_upward
4
arrow_downward

Либо:

a) Используйте класс ограниченной очереди, который блокирует производителя, если размер очереди достигает MAX_THRESHOLD. Это означает изменение класса очереди, которое может вам не понадобиться.

б) Используйте «очередь пула» - еще одну неограниченную очередь блокировки, которую вы заполняете объектами MAX_THRESHOLD при запуске. Производитель получает свои объекты из пула, загружает их, ставит в очередь производителю. Производитель получает объекты от потребителя, «потребляет» их и возвращает в пул. Это своего рода мандат на использование указателей или, возможно, ссылок, которые вам могут не понадобиться.

c) Используйте семафор, инициализированный счетчиком MAX_THRESHOLD, для представления токенов сообщений аналогично пункту (b) - производитель должен получить единицу перед постановкой в ​​очередь, потребитель отправляет единицу, когда это сделано с объектом сообщения.

Я предпочитаю использовать (b).

person Martin James    schedule 10.03.2013
comment
Не могли бы вы подробнее остановиться на вашем втором подходе - очереди к бассейну? - person Pacerier; 14.09.2015