Здравствуйте, watchmaker, Вы писали:

W>Второй тип потоков — перекладыватель в output. Он работает в единственном экземпляре.
W>В бесконечном цикле ждёт событие "в очереди что-то появилось", под мьютексом делает swap со своей локальной пустой очередью, и потом начинает писать из своей локальной очереди все события в файл/консоль без каких-либо синхронизаций.
W>
W>Собственно, это почти всё.
W>Каких-то сложных потоков данных или блокировок нет — всё собирается из стандартных примитивов.
W>В фантастическом случае, если например промежуточная очередь тормозит, то понятно как её можно заменить (уверен, что кто-нибудь посоветует тут поиграть со всякими lock-free очередями).

W>И легко сделать дедупликацию событий, чтобы промежуточная очередь не росла бесконечно и чтобы программа не упала по превышению памяти, если вдруг запись в output застопорится. Так как обрабатывающий поток будет видеть, что его же блок событий с предыдущей итерации ещё не был передан для записи, и может легко сделать агрегацию (опять же, можно это делать не блокируя соседние потоки, если они есть).

Спасибо за такой развёрнутый ответ! Есть над чем подумать.
То есть возможная более простая, понятная и быстрая реализация:
1 поток слушает по epoll соединению, перекладывает в общую очередь под мьютексом, cond var нотифаит другой поток разгребатель после записи. Другой поток не читает их под мьютексом а берёт себе, чтобы уже параллельно не мешая 1му потоку писать. Правильно понял?
А как сделать этот swap? Чтобы быстро взять данные? Т.е. не использовать какую-то очередь, а просто буфер, указатель на который меняем? Для нового же нужно выделить место? malloc с каким-то значением? Но мы же не знаем сколько epoll поток вычитает. Или фиксируем сколько читать этой константой len — malloc(len)? А в случае быстрой работы не хочется дергать системные функции.

Можно, в принципе, тогда заранее определить размер общего буфера, в который пишет 1ый поток. И этот же размер будет у локального буфера у 2ого потока. Общая len. Тогда 1ый поток получив набор дескрипторов сможет записать только сколько влезет в len. Оповестить. А с остальными что делать? Может, получил 10 событий = 10 fd, а в этом фиксированном буфере выделено байт только для 5ти.

Здравствуйте, watchmaker, Вы писали:

W>Второй тип потоков — перекладыватель в output. Он работает в единственном экземпляре.
W>В бесконечном цикле ждёт событие "в очереди что-то появилось", под мьютексом делает swap со своей локальной пустой очередью, и потом начинает писать из своей локальной очереди все события в файл/консоль без каких-либо синхронизаций.
W>
W>Собственно, это почти всё.
W>Каких-то сложных потоков данных или блокировок нет — всё собирается из стандартных примитивов.
W>В фантастическом случае, если например промежуточная очередь тормозит, то понятно как её можно заменить (уверен, что кто-нибудь посоветует тут поиграть со всякими lock-free очередями).

W>И легко сделать дедупликацию событий, чтобы промежуточная очередь не росла бесконечно и чтобы программа не упала по превышению памяти, если вдруг запись в output застопорится. Так как обрабатывающий поток будет видеть, что его же блок событий с предыдущей итерации ещё не был передан для записи, и может легко сделать агрегацию (опять же, можно это делать не блокируя соседние потоки, если они есть).

Спасибо за такой развёрнутый ответ! Есть над чем подумать.
То есть, возможная более простая, понятная и быстрая реализация:
1 поток слушает по epoll соединению, перекладывает в общую очередь под мьютексом, cond var нотифаит другой поток разгребатель после записи. Другой поток не читает их под мьютексом а берёт себе, чтобы уже параллельно не мешая 1му потоку писать.

Правильно понял?

А как сделать этот swap? Чтобы быстро взять данные? Т.е. не использовать какую-то очередь, а просто буфер, указатель на который меняем? Для нового же нужно выделить место? malloc с каким-то значением? Но мы же не знаем сколько epoll поток вычитает. Или фиксируем сколько читать этой константой len — malloc(len)? А в случае быстрой работы не хочется дергать системные функции.

Можно, в принципе, тогда заранее определить размер общего буфера, в который пишет 1ый поток. И этот же размер будет у локального буфера у 2ого потока. Общая len. Тогда 1ый поток получив набор дескрипторов сможет записать только сколько влезет в len. Оповестить. А с остальными что делать? Может, получил 10 событий = 10 fd, а в этом фиксированном буфере выделено байт только для 5ти.