Очередь запросов

[sergek]

Коллеги, тема заезженная, но я для себя не нашел подходящего решения.
Задача - наблюдение и управление устройствами по протоколу modbus rtu, подключенными к преобразователям интерфейса USB/RS-485 (не суть важно, к контроллерам). Доступ многопользовательский, для этой цели создан многопоточный сервер, выделяющий для каждого подключившегося клиента свои потоки, в которых осуществляется подключение к устройствам. Запросы выполняются в синхронном режиме - подключение, запрос/ответ, отключение. Асинхронный режим также есть, но это тема для другого разговора, сейчас мне нужно решить другую задачу.

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

Все работает нормально, если конкурируют запросы разных пользователей. Циклы ожидания организуются в разных потоках, очередь движется.
Но если один пользователь начинает делать несколько запросов через малый интервал (например, быстро щелкает мышкой по кнопке), то, из-за медленной реакции устройства (а она может быть от 80 мс до 800), его запросы ставятся в очередь. В программе это можно сделать путем быстрого нажатия несколько раз по кнопке Run. В результате все запросы завершаются таймаутом (Waiting result 0 of request ..).

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

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

[lit-uriy]

Я бы не стал использовать несколько очередей, да ещё и вложенных.
Если проволока (линия связи / транспорт) одна, то и очередь на доступ к проволоке тоже одна.

[sergek]

Очередь-то, если речь о массиве/списке, одна. При выполнении запроса на сервере выполняется функция с индивидуальными для каждого запроса параметрами. Назначение очереди в том, чтобы отсрочить выполнение этой функции на время, пока не подойдет ее очередь (т.е. пока не закончится предыдущий вызов функции).
Проблема в другом - как организовать ее ожидание. Иного способа, как использование циклы событий для ожидания и оповещения о том, что из этого цикла пора выходить, я не придумал. Поэтому, если в одном потоке подряд несколько раз вызывается функция, циклы ожидания получаются вложенными.
У меня возникает ощущение, что я что-то не понимаю и, соответственно, не то делаю

[Bepec]

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

[sergek]

Я б сказал вы страдаете фигней...

Это да... Все, спасибо, коллеги, тему можно считать закрытой по причине неправильной постановки задачи

[Igors]

Трудновато понять что происходит

Код

C++ (Qt)
// ожидание запроса в очереди. Выполняется в клиентском потоке
int CThreadSafeQueue::waitForQueue(int requestId, int msecs)
{
    CQueueLoop loop(this, requestId);
    return loop.waitForQueue(msecs);
}
 

Какой-то запрос начал крутить свой loop, но запросы идут и идут, и каждый делает waitForQueue, т.е. запускает свой такой же loop. Выходит вложенность все нарастает (каждый новый loop крутится внутри предыдущего). Или я не так понял?

[sergek]

Все так, если запросы в одном потоке, отчего и проблемы.
А если они в разных потоках (от разных клиентов), работает вот это:

Код:

connect(queue, &CThreadSafeQueue::notify, this, &CQueueLoop::notificationSlot, Qt::QueuedConnection);

т.е. при движении очереди происходит выход из цикла в слоте notificationSlot:

Код:

this->exit(1);

[Igors]

Тогда почему не один CQueueLoop (на все запросы одного клиента)? Все данные "когда соскочить" есть, ну может добавить clientID

[sergek]

Согласен, но тогда другие сложности. Я сейчас пытаюсь все упростить.

[lit-uriy]

Если совсем упрощенно, то у меня так выглядит.

Есть объект "Транспорт" описывающий конкретную проволоку RS-485.
Есть объект "Устройство" описывающий ModBus-устройство.
"Устройства" добавляются в "Транспорт".

В объекте "Транспорт" есть слот "добавитьВОчередь(Устройство, данные)", он дёргается из произвольного потока.
У "Транспорта" есть сигнал "естьОбновление(Устройство"), к которому подключены заинтересованные объекты.
По получению сигнала объекты проверяют новые данные "Устройства", например, если мы включали дискретный выход устройства, то  у него можно спросить состояние этого выхода.

примерно такая реализация

[sergek]

Слишком упрощенно. "Транспорт" - для последовательности запросов к modbus-устройствам? Если да, то в нормальном состоянии длина очереди - 0?
"Устройство" - для хранения текущего состояния устройств? Тогда почему сигнал испускает "Транспорт"?
Заинтересованные объекты по сигналу тупо получают данные из "Устройство"?
Кто выполняет запросы к modbus-устройствам?
Если поясните, будут другие вопросы.

[lit-uriy]

"Транспорт" - для последовательности запросов к modbus-устройствам? Если да, то в нормальном состоянии длина очереди - 0?

Да, сидящим на одной проволоке, т.е. в случае нескольких проволок имеем несколько экземпляров класса "Транспорт".
Максимум - 1, т.к. в любом случае задание помещается в очередь и пока вся транзакция не завершится оно из очереди не удаляется (читай далее).

"Устройство" - для хранения текущего состояния устройств? Тогда почему сигнал испускает "Транспорт"?

Да.
Потому что только "Транспорт" знает когда транзакция завершилась, но устройство тоже посылает сигнал (дубль, следом за транспортом); на практике оказалось, что в одной части программы удобнее работать с транспортом, а в другой с устройством.

Заинтересованные объекты по сигналу тупо получают данные из "Устройство"?

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

Кто выполняет запросы к modbus-устройствам?

Транспорт (он владелец объекта QSerialPort).
Он реализует транзакции на шине, но пакеты (Request PDU) для него формирует "Устройство", он передаёт в "Устройство" принятый пакет (Response PDU) и получает назад либо утвердительный ответ либо ошибку. При ошибке работает специальный алгоритм, повторный запрос со счётчиком неудач либо безусловная посылка (заинтерисованным объектам) сигнала ошибки.
Этот алгоритм уникальный для нашего оборудования и мало связан с ModBus-ом.

В текущей реализации ассоциацию адреса устройства на шине ModBus с экзепляром класса "Устройство" хранит "Транспорт", "транспорт" же и посчитывает CRC ну и окончательно формирует Request ADU тоже "Транспорт". Он же и разбирает "Response ADU" и вытаскивает из него PDU.

P.S.
Концепт в принципе хочется переработать, но более "красивого" решения пока не нарисовалось, изобрести ещё один "некрасивый" и лень, и времени нет, и придётся в дальнейшем поддерживать два "некрасивых", что приведёт к каше в голове.

[sergek]

В общем, понятно, хотя смутило "Максимум - 1". Наверное, минимум? А после завершения транзакции, задание должно удаляться, это я имел в виду, говоря про нормальное состояние (точнее - исходное) с длиной очереди 0.

И не совсем понятен транзакционный механизм.

"Потому что только "Транспорт" знает когда транзакция завершилась".

Транзакция завершается, когда получен ответ. С другой стороны, вы говорите, что

"Т.е. подтверждения завершения конкретной транзакции нет (это минус имеющейся реализации)."

Получается, вы ответ устройства не связываете с запросом? А как тогда вы определяете, что шина (проволока) освободилась, и можно передавать в порт новый пакет?

[lit-uriy]

А после завершения транзакции, задание должно удаляться, это я имел в виду, говоря про нормальное состояние (точнее - исходное) с длиной очереди 0.

Да именно так.

И не совсем понятен транзакционный механизм.

"Потому что только "Транспорт" знает когда транзакция завершилась".

Транзакция завершается, когда получен ответ. С другой стороны, вы говорите, что

"Т.е. подтверждения завершения конкретной транзакции нет (это минус имеющейся реализации)."

Получается, вы ответ устройства не связываете с запросом? А как тогда вы определяете, что шина (проволока) освободилась, и можно передавать в порт новый пакет?

Для высокоуровневого кода нет подтверждения что выполнилось его конкретное задание.

Поэтому заинтересованный вынужден анализировать состояние

коротко: транспорт сигналит о том, что он сделал очередное задание. Ему сказали какому устройству что передать, он это сделает как только будет такая возможность и просигналит.

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

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


Дополнение к предыдущему посту с описанием:
Сигнал, посылаемый устройством, который дублирует сигнал транспорта, содержит список изменившихся переменных устройства (аналог тэга в OPC-серверах).
Высокоуровневый код который не следит за сигналами транспорта, а следит за их дублёрами от устройств может фильтровать сигналы, анализируя полученный список переменных. Если изменилась нужная переменная, то как-то реагируем, если нет, то игнорируем сигнал.

[Igors]

Ну не знаю, может и нет необходимости обсуждать "всю" архитектуру, что обычно трудно и ничем не кончается. Что если попытаться "сузить" так

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

Если так то почему бы не решить это в главном цикле событий или в созданном (но одном), крутимся там, варианты

- сигнал "очередь обновилась" - проверяем мой (клиента) запрос первый - выполняем действия (посылку, удаление) и возвращаемся в свой EventLoop.

- сигнал "таймаут" (не дождались), и далее по задаче