многопоточность — блокировка очереди C ++ Segfault w / Boost

У меня была потребность в очереди блокировки в C ++ с возможностью тайм-аута offer(), Очередь предназначена для несколько производителей, один потребитель. Когда я реализовывал, я не нашел хороших существующих очередей, которые бы соответствовали этой потребности, поэтому я сам кодировал их.

Я вижу, что сегфолты выходят из take() метод в очереди, но они с перебоями. Я искал код для проблем, но я не вижу ничего, что выглядит проблематичным.

Мне интересно, если:

• Существует существующая библиотека, которая делает это надежно, что я должен
  использовать (предпочтительнее повышение или только заголовок).
• Любой видит в моем коде очевидный недостаток, который мне нужно исправить.

Вот заголовок:

class BlockingQueue
{
public:
BlockingQueue(unsigned int capacity) : capacity(capacity) { };
bool offer(const MyType & myType, unsigned int timeoutMillis);
MyType take();
void put(const MyType & myType);
unsigned int getCapacity();
unsigned int getCount();

private:
std::deque<MyType> queue;
unsigned int capacity;
};

И соответствующие реализации:

boost::condition_variable cond;
boost::mutex mut;

bool BlockingQueue::offer(const MyType & myType, unsigned int timeoutMillis)
{
Timer timer;

// boost::unique_lock is a scoped lock - its destructor will call unlock().
// So no need for us to make that call here.
boost::unique_lock<boost::mutex> lock(mut);

// We use a while loop here because the monitor may have woken up because
// another producer did a PulseAll. In that case, the queue may not have
// room, so we need to re-check and re-wait if that is the case.
// We use an external stopwatch to stop the madness if we have taken too long.
while (queue.size() >= this->capacity)
{
int monitorTimeout = timeoutMillis - ((unsigned int) timer.getElapsedMilliSeconds());

if (monitorTimeout <= 0)
{
return false;
}

if (!cond.timed_wait(lock, boost::posix_time::milliseconds(timeoutMillis)))
{
return false;
}
}

cond.notify_all();

queue.push_back(myType);

return true;
}

void BlockingQueue::put(const MyType & myType)
{
// boost::unique_lock is a scoped lock - its destructor will call unlock().
// So no need for us to make that call here.
boost::unique_lock<boost::mutex> lock(mut);

// We use a while loop here because the monitor may have woken up because
// another producer did a PulseAll. In that case, the queue may not have
// room, so we need to re-check and re-wait if that is the case.
// We use an external stopwatch to stop the madness if we have taken too long.
while (queue.size() >= this->capacity)
{
cond.wait(lock);
}

cond.notify_all();

queue.push_back(myType);
}

MyType BlockingQueue::take()
{
// boost::unique_lock is a scoped lock - its destructor will call unlock().
// So no need for us to make that call here.
boost::unique_lock<boost::mutex> lock(mut);

while (queue.size() == 0)
{
cond.wait(lock);
}

cond.notify_one();

MyType myType = this->queue.front();

this->queue.pop_front();

return myType;
}

unsigned int BlockingQueue::getCapacity()
{
return this->capacity;
}

unsigned int BlockingQueue::getCount()
{
return this->queue.size();
}

И да, я не реализовывал класс с помощью шаблонов — это следующий в списке 🙂

Любая помощь с благодарностью. Проблемы с потоками могут быть действительно трудно определить.

-Бен