Пользовательская очередь С++ Producer с (очень) быстрой и надежной передачей обслуживания

Привет, я занимаюсь передачей потоков, используя быструю и надежную очередь потребителей-производителей. Я работаю над Windows с VС++.

Я основывал свой дизайн на Anthony Williams, то есть, в основном, boost:: mutex с boost:: condition_variable. Теперь обычно время между notify_one() и пробуждением варьируется от 10 (редких) до 100 микросекунд, причем большинство значений составляет 50 микросекунд. Тем не менее, около 1 в 1000 идет более одной миллисекунды, причем некоторые из них занимают более 5 миллисекунд.

Мне просто интересно, являются ли эти типичные значения? Есть ли более быстрый способ сигнализации, не требующий вращения? Отсюда все зависит от приоритетов потоков? Я не начал играть с приоритетами, но мне было просто интересно, есть ли вероятность получить это в довольно устойчивой области около 10 микросекунд?

Спасибо

РЕДАКТИРОВАТЬ: С помощью SetPriorityClass (GetCurrentProcess(), REALTIME_PRIORITY_CLASS) среднее время пробуждения по-прежнему составляет примерно 50 микросекунд, однако есть намного меньше выбросов, большинство из которых теперь составляют около 150-200 микронов. За исключением одного уродского выброса в 7 мс. Хммм... не хорошо.

Ответ 1

Короткий ответ - да, оттуда это действительно зависит от управления операционной системой и планирования потоков. RTS (системы реального времени) могут довести эти 50 микронов до 15 микронов, и, что более важно, они могут избавиться от выбросов. В противном случае вращение - единственный ответ. Если количество очередей больше, чем ядер, то идея может состоять в том, чтобы иметь x число потоков, вращающихся, реагировать немедленно и оставшуюся блокировку. Это будет связано с каким-то потоком очереди "хозяина", постоянно вращающимся для проверки всех очередей и - либо самой обработкой элементов, либо передачей их рабочим потокам, из которых некоторые также могут вращаться, чтобы сохранить эти 50 микронов. Однако это осложняется.

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

Сложно, но возможно. Если я когда-нибудь его установлю, я также могу опубликовать код.

Ответ 2

Один из способов амортизировать накладные расходы на блокировку и пробуждение потоков - это добавить вторую очередь и реализовать подход с двойной буферизацией. Это позволяет осуществлять пакетную обработку на стороне потребителя:

template<typename F>
std::size_t consume_all(F&& f)
{   
    // minimize the scope of the lock
    {
        std::lock_guard<std::mutex> lock(the_mutex);
        std::swap(the_queue, the_queue2);
    }

    // process all items from the_queue2 in batch
    for (auto& item : the_queue2)
    {
        f(item);
    }

    auto result = the_queue2.size();        
    the_queue2.clear(); // clears the queue and preserves the memory. perfect!
    return result;
}

Рабочий пример кода.

Это не устраняет проблему с задержкой, но может повысить пропускную способность. Если происходит икота, потребитель будет представлен большой партией, которая затем может обрабатываться с полной скоростью без каких-либо блокировок. Это позволяет потребителю быстро догнать производителя.

Ответ 3

Есть много вещей, которые могут вызвать проблемы.

Вам, вероятно, нужно попытаться профилировать приложение и посмотреть, где могут произойти замедление.

Некоторые примечания:

Являются ли потребитель и производитель одним и тем же процессом? Если это так, критический раздел намного быстрее, чем Mutex.
Попытайтесь обеспечить, чтобы вся память очереди находилась в текущей памяти. Если вам нужно поменять местами страницы, которые замедлятся.
Будьте очень осторожны, устанавливая свой процесс на приоритет в реальном времени. Это должно быть для системных процессов. Если процесс слишком много работает, он может предотвратить критический системный процесс, получающий CPU, который может закончиться очень плохо. Если вы не нуждаетесь в реальном времени, просто используйте HIGH_PRIORITY_CLASS