Как лучше организовать многопоточную архитектуру?

Question

[Cyapa]

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

Логика управляющих потоков и коммутатора итерационная, то есть у реализующих классов есть метод tick, который должен вызываться в цикле.

Внимание вопрос: как лучше это все организовать? Есть ли какие-то шаблонные решения (по типу пула потоков)?

Answer 1

[Евгений Шатунов]

Смотри, прямо готового монолитного шаблона пожалуй нет (точнее, я со своего опыта такой не нахожу).
По описанию видно что можно использовать Thread pool + Proactor + Actor model.
Вместо потока на девайс будет пул потоков + управляющий поток, т.е. семейства потоков только два. Каждый девайс - актор. Действия актора выполняются атомарно в необходимом потоке (главный/сервисный) благодаря планированию нужного обработчика для нужного потока.

Вот, подкину литературку:
www.gamedev.ru/code/articles/concurrency_models
https://habrahabr.ru/users/eao197/topics/ - SObjectizer является довольно интересной библиотекой.

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

Comments

[Cyapa]

Никак не могу уловить вашу идею на счет пула потоков. Обычно пул потоков применяется если нужно масштабирование. В данной же задаче оно не нужно, количество потоков всегда равно количеству девайсов. Можете описать поподробнее то, что вы имеете ввиду?

[Евгений Шатунов]

Cyapa , да, могу. :)
https://en.wikipedia.org/wiki/Thread_pool

Пул потоков в данном случае представлен пулом работников для параллельной обработки задач. Задачей является любой сервис (тик, посылка команды, проверка наличия данных в буфере приема, чтение буфера приема и.т.д) одного конкретного девайса.
Число работников в пуле фиксировано, никак не связано с числом девайсов и значительно меньше числа подключенных девайсов (2/4/8 работников против 20+ девайсов). Люди любят делать число работников в пуле по числу ядер процессора, крепить работников к отдельным ядрам и типа наслаждаться слабыми коллизиями планирования.
Помимо работников у тебя еще всегда есть главный поток процесса, который тоже является работником, но выполняет только те задачи, которые нужно выполнить только в главном потоке (например, запуск callback функций после выполнения асинхронных задач).

Сервисом конкретного устройства занимается один конкретный актер (Actor model). Актер занимается планированием задач сервиса устройства (используя схему Reactor/Proactor). Сервисом устройства можно назвать опрос на предмет потери соединения с устройством, пинг устройства, посылку команд на устройство, проверку ответа от устройства, может что-нибудь еще...
Актер устройства планирует действие, которое должно быть выполнено асинхронно одним из работников в потоке работника.

Почему именно Proactor, а не Reactor. Реактор применяется тогда, когда необходимо приостановить работу вызывающего потока на время асинхронной операции. Выглядит это как простой вызов функции. Чтение из файла или сокета в синхронном режиме, для примера.
Проактор применяется тогда, как есть возможность запланировать некоторые действия в определенной последовательности - поставить callback на исполнение после асинхронной операции.
Я думаю, Проактор для тебя может оказаться очень удобным.

Почему я предложил именно такую схему. По описанию я увидел что потоками ты задумал облегчить себе работу. Сделать по потоку на устройство, а внутри потока у тебя типа обычная синхронная программа с синхронным сервисом устройства. Так ведь?
В любом случае это означает что потоки у тебя большую часть времени спят. Именно в этом случае куда удобнее исправить все на предложенную схему. Плюсом, сервис устройств станет удобнее, особенно если разделишь задачи по пингу/проверке живости устройства с задачами отсылки/проверки наличия/приема данных.

[Евгений Шатунов]

Или тебе прям принципиальную схему описать? :)

[Cyapa]

Интересный подход. С радостью бы поглянул на принципиальную схему. Если не сложно, конечно. Ведь дьявол, как известно, в деталях.

[Евгений Шатунов]

Cyapa , дьявол в деталях - это правда. :)
Смотри, в таком случае мне нужна информация. Ты случайно не IoT-хаб проектируешь?
Как узнается что в системе появилось еще одно устройство, с которым надо взаимодействовать?
Как у тебя сейчас сделано, чем именно занимаются потоки для устройств?
У меня еще рабочее время, разница между нами - 5 часов, а такой объем писанины и сторонней прагматики я смогу себе только после работы позволить. :)

[Cyapa]

Речь идет о GSM-модемах.
До узнавания о новом устройстве дело пока не дошло, так как это второстепенная задача. Для Win можно просто подписаться на уведомление о новых устройствах, для Nix пока не смотрел.
Потоки инициализируют девайсы и раз в 5 секунд чекают на предмет новых данных. Если данные есть, скидывают их коммутатору. Между этим делом спят. Вся работа через AT-команды.
Пока просто агрегировал std::thread в рабочих классах. Запускается в конструкторе, джойнится в деструкторе. Бесконечный цикл с флагом вызывающий рабочий метод tick.

Если переходить на предложенную схему, то по хорошему нужно переделать все, даже общение с девайсами.

Ничего, я, конечно же, подожду. :) Заодно будет время все получе обдумать.

[Евгений Шатунов]

А, я уже рисовал себе в уме какую-нибудь гирлянду из десятков устройств на проводах. :)
Побуду немного теоретиком кайфа и накидаю видение этого подхода.

Первое, чем надо озаботиться - это таймеры и их планировщик.
Это простая штука, ее обычно делают на основном потоке, в главном цикле приложения. Такой планировщик задействуется с определенной дискретностью (обычно 50-150 раз в секунду) для активации сработавших таймеров.
Любители особой экзотики или больших нагрузок делают для таймеров отдельную очередь задач для работника из пула потоков (можно сделать так, чтобы задачи из очереди таймеров вынимались только в соответствующее время, а в остальное время - чтобы работник разбирал другие очереди).

Что тут есть пул потоков и работники. Пул потоков предоставляет только управление небольшим количеством потоков. Потоки создаются при инициализации пула и живут до его деинициализации. Работник – это цикл обработки задач. Задачи поступают из очередей. Очередь задач помечается некоторым семейством. Вон, если применять асинхронный планировщик таймеров, то для него понадобится свое отдельное семейство и очередь (и эта очередь должна отдавать задачи только для уже сработавших таймеров). Задача всегда планируется для конкретной очереди.
Для каждого потока из пула потоков создается работник и передается в этот поток. Работнику назначается набор семейств очередей с установленными приоритетами. При наличии задач в очереди, работник забирает не все, уповая на то, что эту очередь обслуживают несколько работников. Обычно из очереди берут не больше N задач. Для очередей характерна синхронизация семафором, т.е. работники просто усыпляются на ожидании задач в случае их отсутствия.
Но есть еще один работник, чье поведение нехарактерно – это работник главного потока. Он не блокирует свой поток при отсутствии задач и для него определяются свое отдельное от всех семейство и очередь задач. Этот работник выполняет только то, что непременно надо выполнить в главном потоке процесса.
Каждую задачу можно сопроводить специальным обработчиком, который автоматически планируется в необходимую очередь сразу после завершения задачи. Во время планирования в этот обработчик передается результат выполнения задачи. Во время планирования задачи в ее очередь, к задаче добавляется этот самый обработчик с пометкой его очереди. Это дает возможность саму задачу выполнить асинхронно, а ее результат, например, обработать в главном потоке. Таким же образом можно организовать цепное планирование задач в зависимости от результатов предыдущих задач в цепи. Это очень простая схема проактора, полноценный проактор нам тут и не нужен.

Теперь - определение наличия нового устройства. Диспетчер новых устройств - это такой же актер, в список дел которого входит мониторинг оборудования и создание актеров для каждого нового устройства. Его полезно сделать актером потому, что может потребоваться сделать мониторинг асинхронным. Именно диспетчер новых устройств можно подписать на уведомление о новых устройствах.

В момент обнаружения нового устройства для него создается актер. Актер устройства планирует себе таймер для проверки живости устройства. В общем смысле это называется ping/heartbeat service. Если подключенное устройство само с некоторым интервалом отсылает сигнал своей живости - это heartbeat, его можно просто читать и обновлять состояние живости. В ином случае надо делать схему ping-pong с интервалом ожидания ответа(pong). Для ping service удобно запланировать два таймера, из которых только один будет включен в одно конкретное время. Ping-таймер отмеряет интервал между отсылкой пинга. Отсылка пинга останавливает Ping-таймер и запускает Pong-таймер. Последний считает интервал ожидания ответа от устройства. Таймеры снова меняются местами в момент получения ответа от устройства или истечения времени Pong-таймера. Именно в этом месте можно реализовать поведение при потере устройства.

Прием данных с устройства делается тоже по таймеру с маленьким интервалом. По этому таймеру планируется только опрос буфера чтения (эмм... у тебя там сокеты?) на возможность чтения.
А фактическое чтение выполняется только в случае успеха первой задачи (т.е. если проверка показала что читать можно).
Это - планирование двух асинхронных увязанных задач (проверка + чтение) с условием, что вторая задача выполнится только если результат первой задачи успешен.
Callback задачи чтения производит диспетчеризацию принятого буфера с помощью работника основного потока.

Для клиента устройства актер и будет этим самым устройством. Клиент использует публичный интерфейс актера для общения с устройством. Сам актер, по запросу клиента формирует или одну задачу, или цепочку задач.
Скажем, надо получить с устройства некоторые данные. У клиента устройства есть хендлер для этих данных. Клиент вызывает функцию актера устройства, передавая туда хендлер. Актер планирует задачи:
1) послать на устройство запрос данных из потока любого свободного работника (вот тут я не спросил, синхронные ли эти запросы);
2) вызвать клиентский хендлер с принятыми данными в момент их диспетчеризации.

Все остальные моменты, пожалуй, лучше уже тебе самостоятельно обдумать.

[Cyapa]

Благодарю за развернутый ответ. Решение действительно интересное. Ту часть, что про асинхронное общение с устройством и коллбэки я уже реализовывал в ранних версиях проекта. Но потом пришлось отказаться от такого, потому как возникала лютая вложенность. К примеру, для инициализации устройства нужно 10-15 последовательных команд. Мешанина выходит страшная. :)
Сейчас у меня уже поджимает время и придется все реализовывать по той схеме, что уже имеется. После этого хочу попробовать реализовать описанный вами вариант, даже не смотря на проблему с большой вложенностью коллбэков. Думаю, по ходу действия у меня возникнет не мало вопросов на счет определенных деталей. Не возражаете, если я, вдруг что, задам их в этой ветке?

[Евгений Шатунов]

Cyapa , да, если будут вопросы, ты можешь или тут их задавать, или оформлять новый вопрос и сразу приглашать меня. :)
Новый вопрос будет даже лучше, чтобы все люди могли поделиться своими навыками.

[Cyapa]

Евгений Шатунов:
Ну вот, спустя без малого год, пришло время переписать проект с нуля. И честно говоря, я уже дня три не могу придумать даже с какой стороны ухватиться за это дело. Не менее 20 раз перечитал ваши комментарии, изучил все прикрепленные ссылки, разобрался с теорией... Но на практике никак не выходит это применить. Как только я начинаю продумывать конкретное применение, так сразу находится множество подводных камней. В связи с этим возникает гнетущее чувство неопределенности: то ли я что-то продумываю не так, то ли этот подход не применим к моей задаче.
В общем, очень бы хотелось получить мнение со стороны. Если вы не против небольшого обсуждения - дайте, пожалуйста, знать когда у вас появится свободная минутка, хорошо?

[Евгений Шатунов]

Cyapa , только что выполз из отпуска. Лето...
Как тебе удобнее обсудить? В частном порядке или в виде открытого форума?
Мой скайп : evgeniy.shatunov

Дополнительно, я тут ознакомился с акторным фреймворком SObjectizer.
Создатели фреймворка пишут о нем на хабре: https://habrahabr.ru/users/eao197/topics/

[Cyapa]

Евгений Шатунов Думаю, в частном порядке будет быстрее.
Может у вас есть телеграм или icq?

Да, я читал про SObjectizer на хабре по вашей строй ссылке.

[Евгений Шатунов]

Cyapa , аська - 2209825

[Cyapa]

Евгений Шатунов отписал.

Answer 2

[Александр Таратин]

Из-за ошибок оборудования управляющий поток может завершить свою работу.

При таком раскладе я бы так вообще по отдельным процессам разнес, чтобы меньше думать о возможном некорректном высвобождении ресурсов.

Comments

[Cyapa]

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