Возможна тупиковая ситуация в C ++ / boost / thread

Question

Предположим, что следующий код выполняется на одноядерном процессоре:

#include <cstdio>
#include <boost/thread.hpp>
#include <boost/thread/condition.hpp>
#include "boost/date_time/posix_time/posix_time.hpp"
#include <deque>
#include <cstdlib>
#include <time.h>

std::deque<int> buffer;
boost::mutex bufferMutex;

boost::condition bufferHasSome;
boost::condition bufferEmpty;

void Reader()
{
    boost::mutex mutex;
    boost::mutex::scoped_lock lock(mutex);

    //read as fast as possible:
    while(true)
    {
        while(buffer.size() <= 0) //1.1
        {
            bufferHasSome.wait(lock); //1.2
        }

        bufferMutex.lock();
        for(int i = 0; i < buffer.size(); i++)
        {
            printf("%d\n", buffer.front());
            buffer.pop_front();
        }
        bufferMutex.unlock();

        //everything was read:
        bufferEmpty.notify_one();
    }
}

void Writer()
{
    boost::mutex mutex;
    boost::mutex::scoped_lock lock(mutex);

    int index = 0;

    while(true)
    {
        //write portion:
        for(int i = rand() % 5; i >= 0; i--)
        {
            bufferMutex.lock();
            buffer.push_back(index);
            bufferMutex.unlock(); //2.1

            bufferHasSome.notify_one(); //2.2

            index++;
            boost::this_thread::sleep(boost::posix_time::milliseconds(rand() % 10));
        }

        //definetely wait while written portion will be read:
        while(buffer.size() > 0)
        {
            bufferEmpty.wait(lock);
        }
    }
}

int main()
{
    srand(time(NULL));

    boost::thread readerThread(Reader);
    boost::thread writerThread(Writer);

    getchar();

    return 0;
}

и процессор остановлен после 1.1 (где size = 0) в потоке Reader и переключился на Writer , где индекс был добавлен ( 2.1 ) в буфер и bufferHasSome был уведомлен (в 2.2 ) (но пока его никто не ждет, так что это была просто недействительная операция);затем процессор переключился обратно на поток Reader и начал (на 1.2 ) ждать, пока кто-нибудь запишет что-то в буфер , но только тот, кто может записать, ожидаеткто-нибудь, чтобы прочитать буфер .Эта программа зависает в среднем после 150 итераций - я думаю, что из-за этого.Что я пропустил?Как это исправить?

Answer 1

Я вижу здесь пару проблем.Самое главное, вы проверяете общие значения (а именно, buffer.size ()) вне блокировки.Во-вторых, у вас есть эти странные мьютексы, локальные для каждой функции, которые абсолютно ничего не делают, поскольку они не разделяются между потоками.Если вы заблокируете bufferMutex перед проверкой buffer.size (), затем подождите на bufferMutex (то есть вы разблокируете его, что правильно, а затем снова заблокируете его, когда поток уведомлен), я думаю, что угроза взаимоблокировки должна исчезнуть.

Answer 2

Не совсем ответ на ваш вопрос. Другие, кажется, начали с этого. Я хотел бы предложить вам взглянуть на concurrent_bounded_queue в TBB . Если вы используете это, ваш код будет упрощен и менее подвержен ошибкам.

#include <cstdio>
#include <cstdlib>
#include <time.h>

#include <boost/thread.hpp>
#include <boost/date_time/posix_time/posix_time.hpp>

#include <tbb/concurrent_queue.h>

tbb::concurrent_bounded_queue buffer;

void Reader()
{
    while(true)
    {
        int value;
        buffer.pop(value); // blocks when empty
        printf("%d\n", value);
    }
}

void Writer()
{
    int index = 0;

    buffer.set_capacity(5); // If buffer.size() > 5 then push will block.

    while(true)
    {
        for(int i = rand() % 5; i >= 0; i--)
        {
            buffer.push(index++); // blocks when full
            boost::this_thread::sleep(boost::posix_time::milliseconds(rand() % 10));
        }

        // "definetely wait while written portion will be read."
        // Not sure what exactly the purpose of this was. But I guess set_capacity will fulfill the same role.
    }
}

int main()
{
    srand(time(NULL));

    boost::thread readerThread(Reader);
    boost::thread writerThread(Writer);

    getchar();

    return 0;
}

Answer 3

Предположительно, ваш писатель ждет, пока читатель закончит читать, прежде чем снова писать.

В любом случае вам нужен только один "глобальный" мьютекс, и вы должны использовать его также при ожидании переменных состояния.

Локальные мьютексы не действуют.

Я бы также предложил, чтобы ваша основная функция соединяла ваши два потока (я бы использовал thread_group). Вам также понадобятся некоторые завершающие условия для вашего цикла. Может быть, писатель установит переменную после ее завершения и рассылки, а ваш читательский поток проверит это условие, а также проверит состояние очереди.

Answer 4

Ваша проблема может быть связана с этим циклом чтения:

    for(int i = 0; i < buffer.size(); i++)
    {
        printf("%d\n", buffer.front());
        buffer.pop_front();
    }

Проблема в том, что выше будет читать только половину элементов. Поскольку по мере того, как вы выталкиваете элементы, buffer.size () уменьшается, поэтому итерации заканчиваются, когда размер в два раза меньше того, который был при его запуске. Вы должны просто заменить его циклом while:

    while(buffer.size() > 0)
    {
        printf("%d\n", buffer.front());
        buffer.pop_front();
    }

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