Блог

Я написал очередь блокировки на C, очень похожую на ту, которая предоставляется классом BlockingQueue Java. Очередь в основном предоставляет следующий API:

 void blocking_queue_add(Blocking_Queue* bq, void* element);
void* blocking_queue_get(Blocking_Queue* bq);
 

и обладает следующими свойствами:

• Она имеет фиксированный размер.
• Если blocking_queue_add вызывается, когда очередь заполнена, вызывающий блокируется до тех пор, пока в очереди не останется свободного места.
• Если blocking_queue_get вызывается, когда очередь пуста, вызывающий блокируется до тех пор, пока не появятся данные для извлечения.
• Если несколько вызывающих абонентов заблокированы в ожидании blocking_queue_add или blocking_queue_get , их необходимо обслуживать по порядку (FIFO).

Мне трудно реализовать очередь, которая удовлетворяет последней точке при сохранении приемлемой производительности. Я использую pthreads для синхронизации и, чтобы удовлетворить требование FIFO, я полагаюсь на pthread_cond_wait pthread_cond_broadcast функции and . Вся идея заключается в том, что все заблокированные вызывающие абоненты будут ожидать выполнения условия. Как только один из них может быть обслужен, выдается широковещательный сигнал, и все они просыпаются. На этом этапе они могут легко проверить, кто выбран, на основе внутреннего состояния FIFO. Остальные добровольно блокируют себя, вызывая pthread_cond_wait снова, и ждут своей очереди.

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

Я ищу подсказки о том, как это решить. Я сравнил свою очередь с реализацией Java, и она работает быстрее при работе с небольшими рабочими нагрузками, но когда я тестирую сценарий с большой и полностью несбалансированной рабочей нагрузкой, моя реализация — катастрофа, а реализация Java, похоже, только линейно медленнее. На самом деле, если я удалю весь код, связанный с очередью, из своей реализации и просто изменю обе blocking_queue_add blocking_queue_get функции and, чтобы просто использовать логику cond / broadcast , это уже катастрофа.

Приветствуется любой ввод, спасибо.

Идея Брэндона значительно улучшила производительность. Производительность теперь, похоже, более чем в 2 раза выше, чем в реализации Java.

Для тех, кто интересуется будущим, я сделал исходный код открытым исходным кодом: https://github.com/felipeek/c-fifo-blocking-queue

мне трудно реализовать очередь, которая удовлетворяет последней точке при сохранении приемлемой производительности. Я использую pthreads для синхронизации и, чтобы удовлетворить требование FIFO, я полагаюсь на функции pthread_cond_wait и pthread_cond_broadcast .

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

Чтобы исправить это; каждому потоку нужна своя отдельная переменная условия. Когда поток начинает ожидать данных из пустой очереди, он помещает свою переменную условия в «очередь ожидающих читателей»; и когда поток добавляет данные в очередь, он берет первую переменную условия из «очереди ожидающих читателей» и (если есть официант) выполняет одно pthread_cond_signal() (ине широковещательная передача), чтобы разблокировать только один ожидающий поток.

Обратите внимание, что «очередь ожидающих переменных состояния читателя» может быть связанным списком « struct waiter { struct waiter * next; pthread_cond_t condVar; } » структур; и эти структуры могут создаваться и инициализироваться при создании потока, а затем постоянно перерабатываться (пока поток не завершится).

Для «нескольких авторов» это, по сути, одна и та же проблема с одним и тем же решением (и может повторно использовать тот же « struct waiter «, созданный при создании потока). Когда потоку нужно подождать, чтобы добавить данные в очередь, он добавляет свою переменную условия в «связанный список ожидающих авторов», а когда поток заканчивает удаление данных из очереди, он делает pthread_cond_signal() это, чтобы разблокировать следующего ожидающего автора.

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