Использование мьютексов и условных переменных

Question

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

Существует два потока: писатель и читатель, которые используют общую память. Именованный мьютекс используется для синхронизации доступа к объектам (строковым и векторным) в разделяемой памяти. условная переменная, используемая для ожидания, когда читатель запишет данные в общую память. Таким образом, сценарий таков: - читатель запускается и инициирует условную переменную на именованном мьютексе с условием, что вектор с данными должен быть непустым. и ждет... - писатель блокирует мьютекс и заполняет вектор - писатель "уведомляет один", что запись в вектор данных завершена, и разблокирует мьютекс - читатель получает уведомление, блокирует мьютекс и обрабатывает данные в векторе.

после этого читатель должен уведомить автора о том, что чтение завершено, и вектор может быть снова заполнен новой порцией данных.

поэтому я не уверен, как правильно настроить все эти ожидания и уведомления. Похоже, моя версия зашла в тупик. Пожалуйста, порекомендуйте.

код нити читателя

    namespace bi = boost::interprocess;
    using bi_char_vector = bi::vector<char, CharAllocator>;

    bi::named_mutex             mtx{bi::open_or_create, "mtx"};
    bi::named_condition         cnd{bi::open_or_create, "cnd"};

    data = segment.find_or_construct<bi_char_vector>("data")(segment.get_segment_manager());
    while (!done) {
       bi::scoped_lock<bi::named_mutex> lock{mtx};
       cnd.wait(lock, [data] {return !data->empty(); });

       // process the data...

       cnd.notify_one();
    }

авторский код потока:

    bi::managed_shared_memory segment(bi::open_only, shm_name.c_str());
    bi::named_mutex mtx{bi::open_only, "mtx"};
    bi::named_condition cnd(bi::open_only, "cnd");

    data = segment.find_or_construct<bi_char_vector>("data")(segment.get_segment_manager());

    for(std::size_t chunk_num = 0; chunk_num < chunk_count; ++chunk_num) {
        bi::scoped_lock<bi::named_mutex> lock { mtx };
        cnd.wait(lock);
        data->clear();
        // fills the data
        cnd.notify_one();
    }
}

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

Answer 1

Нашел проблему. на стороне читателя добавлено:

while (!done) {
   bi::scoped_lock<bi::named_mutex> lock{mtx};
   cnd.wait(lock, [data] {return !data->empty(); });

   // process the data...

   data->clear();
   cnd.notify_one();
}

на стороне писателя:

    for(std::size_t chunk_num = 0; chunk_num < chunk_count; ++chunk_num) {
        bi::scoped_lock<bi::named_mutex> lock { mtx };
        cnd.wait(lock, [data] {return data->empty(); });
        //...
        cnd.notify_one();
    }

и теперь работает как положено