Асинхронна комуникация, Няколко устройства на обща шина

Имам странна постановка при която трябва да синхронизирам 3 нишки (четене, писане и клиент).

От едната подавам команда и чакам отговорът асинхронно използвайки подобен интерфейс (има трансформации в последствие, но това е основата):

buffer се подава по серийна шина на която има свързани различни устройства. Те генерират събития и същевременно отговарят и на команди. Когато генерират събитие, в него има адрес на устройството (може да се определи източникът му). При отговор на командите обаче това е невъзможно (няма данни които да го определят еднозначно).

Постановката е следната: Една нишка чете входящите данни (и ги маршрутизира според типът). Основната цел на тази нишка е да лови събитията но се налага тя да обработва и отговорите на командите (където е проблемът). Всяка команда има адрес с който се посочва получателя и той отговаря на нея.

Ето и основния проблем

две команди се пускат по шината (към различни устройства). Въпреки, чв командите се подават синхронно, една след друга по шината, то всяко устройство може да отговори по ралично време (няма гаранция кое ще отговори първо). Това налага send да поставя командидте в опашка която да се чете от друга нишка, която да изпраща данните и да блокира докато чака отговор. Едва след като първата команда получи отговор нишкката коиято изпраща данните ще вземе следващата команда от опашката и ще я изпрати по шината. За да работи тази постановка Future обектът свързан с всяка командата трябва да се споделя между девте нишки.

И точно тук е проблема. Какво да ползвам за целта на споделянето и съответно синхронизацията.

Exchanger<CompletableFuture<ByteBuffer>> върши иделна работа (блокира изпращащата нишка докато четящата не поиска Future обектът когато получи данни). Проблемът обаче е, че ако при четене се получи нещо което се определи за отговор на команда (а не е) то четящата нишка ще блокира (ще чака Future). Дори да задам timeout, то ще изгубя данни докато нишката е блокирала.

Друг вариант е AtomicReference<CompletableFuture>> при него мога да блокирам изпращащата нишка с get(). Така при получаване на фалшиви данни няма да блокирам а просто няма да има Future обект който да завърша.

Кой вариант е по-удачен, или може би ще предложите друг подход?

Това мнение е било редактирано от thrawn на 24-02-2026, 19:33

На входа пакетите се разделят на събития и отговори. Събитията се записват в опашка и се обработват когато може. Отговорите изпращат нотификация към нишката която изпраща комади. Тази нишка има опашка и ги изпраща една по една и чака отговор, като се получи отгор може да има колбек или нещо подобно, ако не получи отговор връща таймаут и продължава със следващата команда.

Нямам идея как е в Джава аз използвам boost основно, а за ембедед FreeRTOS, където всеки таск си има stdio/stdout/stderr, а към тях пращам през _write/_read syscalls функциите и всеки таск си използва printf/getstr и това е. Кода е еднакъв за всяко устройство.

Това мнение е било редактирано от ici на 24-02-2026, 21:51

Ужас, ужас...
В тая джава няма ли нещо като select/epoll/kevent/iocp, че да не се налага ползването на една камара нишки...

Но и да няма, за какво е нужна нишка за *запис*?
Спиш и чакаш данни, когато имаш нещо за пращане - слагаш го в опашката и просто събуждаш нишката, която *чете*. Тя си решава дали сега може да изпрати нещо или в момента дъвче входящи данни. Като получи събитие, праща сигнал на клиент нишката, като получи отговор на команда - маха командата от опашката и взима следващата, ако няма следваща - продължава да спи и да чака събития.

Плюс всяка команда в опашката има callback който да получи отговора.
Така нишките стават две - UI и worker.

Така е, като няма читав event loop...

В момента съм го направил горе долу така. CompletableFuture в java е нещо като Promise в javascript. В примерът thenApply приема callback който ще се изпълни само ако пристигнат данни.

Самото писнае (изпращането на данни) следва този алгоритъм (в отделна нишка)

Друга нишка която само чете (частта е само груба обработка на отговор на команда)

Тук двете нишки си комуникират и се синхронизират с atomicReference като изпращащата нишка не блокира ако липсва response обект (или ако има вече приключен такъв). В кода може да има грешки, че го пиша директно тук. той е само за илюстрация на логиката


@dvader да, те са 2 нишки + UI та 3

Не знам дали стана ясно. Нишката за писане е нужна заради второто блокиране. Трябва да се гарантира, че няма да се изпрати втора команда, докато първата не е приключила. Но не трябва да се блокира UI-a. Той трябва да може да подава команди по всяко време към всяко устройство по шината.



Това мнение е било редактирано от thrawn на 25-02-2026, 09:09

Защо не използваш семафори които да показват заетост на съответното устройство и преди да праща към съответното уствойство да проверява дали е свободно чрез семафора.

Интересна тема. Аз реализирам по Линукс услуги на php които работят по сериен порт с modbus rtu протокол. За сега само с емулатор..

Ами не блокирай де...
Няма ли начин да се пращат сигнали в тоя котлин?
Сигнал, че имаме данни за четене, сигнал, че сме готови да пращаме и т.н.

Не блокира UI а (клиентската нишка). Така може да се поставят няколко команди в опашката (все пак това и е идеята). При стандартна комуникация това е желаното поведение при което данните се обработват синхронно. Имаш команда пращаш я, получаваш отговор и го връщаш. Тук обаче има няколко устройства. Пускаш командата към едното а други две си пращат разни събития (full duplex но с няколко устрийства). Това налага отделната нишка за четене и callback нотификяции за отговори на команди и друг за събитията.
До тук ОК.

Нишката която праща данни вижда команда, буди се, праща я и заспива, докато чака друга команда. Идва втората команда, буди се и я пуска и нея по шината (но тя е за друго устройство). То обаче реагира по-бързо и отговаря първо (въпреки, че командата за него е пусната втора). Това от своя страна налага изпращащата нишка да блокира отново, за да изчака отговорът на всяка команда която изпраща (в потивен слчай, не е ясно къде да се изпрати отговорът).

Пак казвам, няма как да се определи кое устройство е отговорило (отговор на коя команда е). Това е възможно само при събития изпратени от самите устройства.

Ако шината се прекъсне и всяко устойство се пусне да работи самостоятелно проблеми няма да има, защото е ясно, че получения отговор е на последната изпратена кманда.

Между другото, горната схема вече работи доста стабилно.

//off товае java не котлин, не че има значение де

Е, нали пращаш САМО към едно устройство, а събитията можеш да ги отделиш от отговорите? Щом си изпратил команда към едно устройство, то и отговора ще е от него.