Очень многопоточная архитектура приложения.

[RedDog]

Есть некоторое приложение, с большим количеством разнообразных логик: БД, сеть, бизнес-логики N штук, логирование 4 вида.
До некоторого времени разрабатывалось по принципу: "в любой не понятной ситуации создавай отдельный QThread и суй в него свою очередную логику".
Итого, имеем: на БД создается в районе 50 нитей (много подключений, каждое в своем потоке), на сеть около 10 (здесь пул потоков, одна нить несколько подключений обслуживает), вся остальная логика еще на полтинничек тянет.
Делалось это с той целью, что бы одна "логическая единица" не мешала и не лочила работы других, потому как сервер и должен обслуживать клиентов в почти онлайн-режиме.
Вопрос вот в чем: стоит ли продолжать в таком духе и когда пора остановится?
Про пулы потоков в курсе, но есть сложности в определении свободной нити, что бы не лочить при тяжелых вычислениях.

[RedDog]

Я извиняюсь, может вопрос не так задал?
Или все же нормальная практика делать для отдельной логики отдельную нить?

[ssoft]

Я извиняюсь, может вопрос не так задал?
Или все же нормальная практика делать для отдельной логики отдельную нить?

Какой ответ Вы хотите получить?
Если программа работает и хорошо себя чувствует, то, возможно, такой подход оправдан, так как прост и относительно надежен. Это с практической точки зрения.
Если смотреть с технической или академической точки зрения, то чем больше потоков, тем хуже себя чувствует диспетчер ОС.
Если у Вас задачи связанны с высокой производительностью и большими нагрузками, то имхо такой подход неприемлем, лучше посмотреть в сторону корутин буста, на std::future или еще куда-нибудь.
В общем, ответ зависит от критериев разумности в рамках решения конкретной задачи.

[vic57]

Я извиняюсь, может вопрос не так задал?
Или все же нормальная практика делать для отдельной логики отдельную нить?

сервер по любому должен крутиться, а для БД может лучше async?
https://xydan.livejournal.com/8595.html

[RedDog]

Какой ответ Вы хотите получить?
Если программа работает и хорошо себя чувствует, то, возможно, такой подход оправдан, так как прост и относительно надежен. Это с практической точки зрения.
Если смотреть с технической или академической точки зрения, то чем больше потоков, тем хуже себя чувствует диспетчер ОС.
Если у Вас задачи связанны с высокой производительностью и большими нагрузками, то имхо такой подход неприемлем, лучше посмотреть в сторону корутин буста, на std::future или еще куда-нибудь.
В общем, ответ зависит от критериев разумности в рамках решения конкретной задачи.

Пофантазируем...
К примеру есть 8 нитей (QList<QThread>). Есть 32 объекта (SomeWorker), которые должны обрабатывать свои слоты, не мешаясь друг другу.
В начальный момент времени я по какой то логике на каждую нить вешаю по 4 объекта.
Рандомно объекты начинают нагружаться.
Настает момент времени, когда на одной нитке объекты начинают мешать друг другу, следовательно какой то из объектов желательно в соседнюю перекинуть.
Задача в том, как найти наименее нагруженную нитку?
Пока смутно маячит решение в SomeWorker делать а-ля флажок (в каждом слоте?) работает конкретный экземпляр или "курит".
Тогда зная весь массив SomeWorker-ов можно пробежаться и найти нитку с наименьшим/наибольшим кол-вом работающих.
Можно ввести "веса" нагрузки для каждого SomeWorker-а.
Тогда берем наиболее загруженный поток, выдергиваем из него N SomeWorker-ов и кидаем в наименее загруженный.
Ну вот как то так...
М.б. у кого то еще какие мысли будут.

[RedDog]

а для БД может лучше async?

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

[Авварон]

Идея такая - есть несколько рабочих потоков (пусть будет 4) разгребают сообщения. Если обработка занимает мало времени, то они обрабатывают напрямую и кидают сигнал дальше.
Если обработка долгая, то ставим задачу в тред пул, вешаем вотчера, асинхронно ждем сигнала от вотчера.
В целом, можно любую задачу класть в пул, но будут накладные расходы на задачах, которые проще обработать так.
Можно накрутить свой фреймворк поверх QFuture/QFutureInterface который бы исполнял задачу в том же потоке в зависимости от "веса" и сделать это универсально. Идея такая.

[RedDog]

Идея такая - есть несколько рабочих потоков (пусть будет 4) разгребают сообщения.

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

[Авварон]

void run(std::function<void()>) да поможет вам

[Igors]

В начальный момент времени я по какой то логике на каждую нить вешаю по 4 объекта.
Рандомно объекты начинают нагружаться.

А просто очередь задач (возможно с приоритетом)? При поступлении задачи кладете ее в очередь и сигнальчик всем ниткам. Каждая нитка достает задачу их очереди и выполняет ее, Нет задач - возвращается в run до след сигнала

[RedDog]

А просто очередь задач (возможно с приоритетом)?

Будете смеяться, но такая логика уже есть, и она сидит в своем потоке, и шлет сообщения остальным.

PS: вся сложность, что проект довольно большой, около 2000 цпп файлов, и рефакторинг, как я вижу, надо проводить "по верхам", т.е наследоваться от QThread и еще некоторый враппер объектов, которые по потокам раскидываются.
Есть некоторые мысли на этот счет, если вырастут в некий прототип, то выложу на рассмотрение.

[ssoft]

Есть некоторые мысли на этот счет, если вырастут в некий прототип, то выложу на рассмотрение.

Если требуется рефакторинг по "верхам", то прототип тут уже известен - Worker'ы вместо Thread'ов, работающие в пуле потоков.
Оставим Worker'у свою очередь сообщений, добавим признак, что сообщение отправлено в очередь пула потоков для обработки.
Далее такая логика с соответствующими блокировками:

• При добавлении нового сообщения Worker'у, если признак обработки сообщения выставлен, то складываем сообщение в очередь Worker'а, если нет - формируем задание на выполнение и помещаем прямиком в пул потоков. Выставляем признак, что сообщение в обработке
• Выполняем задание.
• После завершения обработки очередного задания пытаемся получить у Worker'а сообщение для формирования нового задания, если есть - помещаем новое задание в очередь пула (или сразу выполняем, зависит от выбранной стратегии)

Такой подход обеспечит отсутствие параллельной работы в рамках каждого Worker'а (за исключением части с очередью сообщений), что здесь и требуется.

ОДНАКО!
Большинство средств Qt реализованы таким образом, что работать с экземплярами объектов можно только в том потоке, с которым ини связаны!
Это относится к сети, GUI, м.б. к SQL и т.п. Поэтому вышеописанный подход напрямую не заработает, придется костыли выдумывать и т.п.

Если ПО нормально функционирует и обеспечивает необходимые характеристики, то мой совет - не нужно ничего менять.
Добавляйте новый функционал по новым правилам, а старый модифицируйте только по мере необходимости.

[RedDog]

Попутный вопрос возник:
Есть некий WorkerMain живущий в своем QThread-е, подписанный на 100500 различных других Worker, которые тоже по ниткам раскиданы.
Необходимо циклично обрабатывать сигналы Worker-ов, таким образом, что бы не было оверхеда в WorkerMain.
Для этого берутся все Worker-ы и испускают по одному сигналу.
Следующий сигнал они не могут испускать, пока очередь сообщений WorkerMain не станет пуста.
Вопрос вот в чем: как достучаться до очереди сообщений QThread-а и узнать, есть ли там что, или она пустая?

[ssoft]

Можно использовать Qt::BlockingQueuedConnection при соединении сигнала Worker со слотом MainWorker.
Тогда в очереди будет не более одного сигнала Worker одновременно.

[RedDog]

Вопрос решился следующим образом:
переопределил eventFilter и в нем при событии QEvent::MetaCall делаю инкремент на входящие/исходящие сообщения

В псевдокоде это выглядит примерно тами образом:

Код:

bool MainWorker::eventFilter(QObject *obj, QEvent *event)
{
    if (event->type() == QEvent::MetaCall)
    {
        QMetaCallEvent* metaEvent = static_cast<QMetaCallEvent*>(event);
        if (nullptr != dynamic_cast<const InputWorker*>(metaEvent->sender()))
        {
            ++m_queueCounter;
        }
        else if (nullptr != dynamic_cast<const OutputWorker*>(metaEvent->sender()))
        {
            if (--m_queueCounter == 0);
                emit imFreeForJob();
        }
    }
}