От известно време ползвам mqtt клиента който идва с новия LWIP (2.0.3):
http://www.nongnu.org/lwip/2_0_x/group__mqtt.html
Всичко му е на шест, работи чудесно. Но днес ударих на една специфична ситуация. Да уточня че тоя клиент ползва RAW API-то за tcp.
Специалното идва когато искам да получа отвън голямо парче данни - т.е. абонирал съм се за определн топик и някой отвън публикува там и брокера ми праща този голям payload. Да кажем че ми стрелват направо файлче от около 1-2 мегабайта.
Сега, чуденката ми е свързана с flow control-а в този момент. Значи, mqtt не прави собствен флоу контрол а разчита на този от tcp. Съответно изобщо не искам да мисля в посока да организирам нещо на ниво mqtt от сорта публикуващия да праща малко, да чака да му публикувам в друг топик отговор, че тогава да праща пак - т.е. не искам на ниво "отгоре над mqtt" да правя флоу контрол.
Та в тази ситуация публикуващият (да кажем PC-то за по-лесно) изстрелва мегабайта файлче и започва да цъка tcp протокола - минават етернет феймчета и от другата страна, в ембедед процесорчето, почвам да получавам данните малко по-малко.
Дотук добре - mqtt клиента от линка горе ми вика callback и ми казва че идват данни - примерно 1000 байта. В колбека тези данни ги взимам и почвам да ги обработвам - да кажем че ги пиша във файл. Самата операция по писане във файл е по-бавна и затова се случва асинхронно - т.е. взимам тия новите 1000 байта и ги давам на таска който ще ги пише да ги тъпче. Т.е. трябва да върна от колбека много преди данните да са записани.
Тъй като данните са копирани за да си работи записващия таск всичко изглежда розово, но скоро ще покафенее... завършвайки колбека аз уведомявам mqtt клиента че съм приключил с това парче данни (но да не забравяме че логически още не съм ги преработил понеже се точат в записващия таск).
Mqtt клиента съответно казва на tcp слоя че работата е свършена и tcp праща ack на брокера че е готов за следващите 1000 байта. Брокерът праща, идва при мен, и почва да се трупат данни докато не се напълнят буферите на стека. Реално стека почва да се мъчи да изхвърли данните към mqtt слоя и вика пак и пак колбека казвайки "ей ти още 1000 байта".
Колбекът (в онази имплементация) няма механизъм да "паузира" - нито мога да блокирам в него, нито мога да върна "чакай малко". Което значи че нямам начин да използвам ефективно (или изобщо) flow control-а на tcp нивото.
За да може това да работи ми трябва по-ефективен достъп до tcp нивото - там има "tcp_recved()" функция която маркира кога горното ниво (т.е. mqtt и нагоре) е приключило с данните. Т.е. кога да се прати ack на tcp ниво и кога брокера/PC-то да прати следващия пакет. В оригиналния код тази "tcp_recved()" се вика даже преди да се повика mqtt колбека, т.е. твърде рано (според мен). Идеята ми е да я махна оттам и да предвидя отделна API функция на mqtt.c която да "довършва" с пращане на ack (викане на tcp_recved) когато дойде отговор от фаиловото писане че е готово (или поне че има място е неговия си буфер).
За да мога да ползвам flow control-а трябва да мога да забавя tcp ack-то до момента в който писането във файла каже че логически тия данни са напълно приключени. На по-горните нива над mqtt-то имам механизмите които връщат това събитие и точно показвам момента, в който да се пусне ack-то навън. Но конкретната имплементация на mqtt-то там няма тази екстра и ще трябва да я добавя.
Може би няма да е проблем ако има начин tcp нивото (в стека) да си буферира няколко пакета - мисля че при мен са конфигуриране 8 PBUF-а. Това няма да попречи, даже ще даде повече време някъде. Но то това ще е в началото - след 8-мия ще се наложи да се бави ack-то докато се освободи поне един буфер.

Та това ми е контекста, както и виждането за посока за решение. Много ли съм се отнесъм в драките? В конкретния случай имам ресурс да го буферирам целия тоя файл от няколко мегабайта в РАМ, но изобщо не искам да правя такива заобикалки - т.е. тази опция не искам да ползвам.
Да поясня че имам RTOS, и съответно стека и "консуматорите" на данните (писането във файл от примера горе) вървят в отделни тредове и си комуникират асинхронно със съобщения.

Едит: ето го мястото в кода:


Вижда ми се че това трябва да се изнесе в отделна фунцкия за да се повика по-късно:

Което май ще позволи да се ползват без копиране данните от PBUF чак във "фаиловата" задача .... (zero copy?)

не си ли се улял с тоя мегабайт
аз правя подобно... с "хитрост"
"лузера/брокера" праща към SUB/клиента линк към файла и си го точа в друг таск с HTTP/FTP
после пращам PUB към базата че файла е done...

Варианти на приложно ниво да заобиколя има много, само дето не виждам смисъл в тях - утройството предлага услуга за нещо си, да я активирам по mqtt и после да я "довършвам" през http е класическото решение, за ъпдейт например. Но то има куп недостатъци - и устройството, и клиента трябва да правят и ползват втори протокол. Клиентът ще трябва да конфигурира някъде сървър, да качи някак файла там, да махне после и т.н.
Ако ще търсим олекотяване няма изобщо да слагам mqtt - може всичкото да се извърти през http. Номерът е че клиента може да има само броузър на телефона някъде на ма..., по света, и оттам през websocket до брокера искам да си свърша работата. Иначе трябва да прехвърлям някъде през облака http заявките, което никак не ми се вижда смислено.

gicho, много нещо си изписал и не знам дали съм ти схванал мъката, ама нещата работят по следния начин:
1) тцп стека има прозорец, който другата страна ще се опита да напълни, преди да поиска ACK. На всеки получен пакет вика receive callback.
2) Желателно е в този callback да не правиш нищо освен да запомниш данните. Особено ако lwip-то ти се търкаля в отделна нишка, ти в тая нишка най-добре да не мажеш. Правиш си примерно някаква опашка, слагаш указател към буферчето и се омиташ. Всъщност аз имам две опашки една за receive и една за accept, но това са подробности.
3) Клиентът, или клиентите си дърпат колкото и когато си искат. При всяко дърпане така или иначе следиш размерите на pbuf-a и като се източи трябва да му извикаш tcp_recved(), желателно в контекста на lwip нишката (не помня задъжително ли беше).


Тва е... не виждам къде ти е дилемата. В крайна сметка скоростта на трансфера ще ти зависи от клиентите с каква скорост празнят опашката. Може да са мнооого бавни, няма проблем, lwip-то ще приеме колкото ти е размера на джама, пакетчетата ще те чакат в опашчицата колкото трябва. Нямаш грижи.

Правилно си ме разбрал и точно момента на викане на tcp_recved() е разковничето. Т.е. за да сихнронизирам бавния консуматорски процес трябва от него да дам сигнал кога да се извика tcp_recved(). Т.е. дали и колко е буфера между tcp таска и приложението не е от критично значение - важното е да имам нотификация обратно от приложението (или буфера ако има такъв) кога са изконсумирани данните (или кога има свободно място в буфера, което е същото), което да послужи на tcp таска да повика tcp_recved() и да затвори верига за Ack до пращащия от другата страна на жицата.
В текущата реализация на mqtt клиента (тази която е включена в 2.0.x версиите на lwip) няма механизъм с който да се "забави" tcp_recved() - той се вика още ПРЕДИ mqtt да извика колбека с който да уведоми навън че има нови данни. Т.е. без модифициране на mqtt клиента там нещата няма как да сработят - така поне го виждам аз?

Не ти знам подробностите, но lwip стека е гъвкав и позволява различни имплементации. Това което трябва да съобразиш е:
1) контекстите на нишките - кое от къде се вика...
2) как се заема паметта на пакетите, щото може да го конфигурираш с локално масивче от буфер, може динамично и т.н.


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

Сега дали трябва да пипаш клиента - не знам, твоя работа... Но пак ти казвам, ако искаш да се бавиш то не бива да се вика tcp_received(), особено в калбака и то преди да си си обработил данните

сега видях че си сложил и код...

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

Не ми се навлиза в подробности, но приеми че изобщо идеята да ползваш такива огромни пакети в MQTT не е добра. Раздели го на няколко части да ти е мирна главата.
Аз лично съм си ограничил моята собствена библиотека до 65 КB (16 бита). Ако пробваш с повече затваря връзката.
Малко недомислен е този протокол. Няма начин да кажеш на сървъра че си зает в момента и да не праща нищо например.

Miro, да, кодът е малко страннен, но не съм съгласен с заключението - викането на колбека е ок и е достатъчно гъвкаво. Имам предвид че е адаптивно към останалата система:
- ако имам ОС мога в колбека да копирам и да пратя съобщение към другата задача, като това може да е оптимизирано за дадения ОС/платформа
- ако нямам ОС и имам кратка обработка - в смисъл ако "затворя очите" и кажа "макс ХХХХ байта щото имам толкова буфер" пак работи и не изисква да правя галибалисткити и накрая пак да ходя на ОС/шедюлър
Т.е. твърдението че е некоректно - ти даваш ЗАЯВКА за обработка и сигнал че твоята част в тази посока е готова - т.е. събитие. Твоя е свободата да обработиш заявката веднага или в друг контекст. Тук явно спорим за това какъв интерфейс за предаване на заявката навън е по-добър - ти предполагам казваш вместо колбек да е някакъв ОС примитив за синхронизация (мейлбокс,ххх). Аз казвам че там е колбек (функцията е изначалната единица код в C) и ако искаш в този колбек закачаш специфичния механизъм - без ос, мейлбокс, фифо, фифо_v2, ....
Бонусът е че кода на клиента (на mqtt) остава непроменен. Пак казвам, то и затова отворих темата, че конкретно там имат дупка в тази имплементация и трябва да се разшири - работя по въпроса в момента Генерално lwip е чудесно организирам и има перфектното API за да покрие подобни дупки и да се адаптира към всяка ситуация.

@Никола - не съм съгласен. Да разделя на няколко части значи да направя/приложа допълнителен протокол, това предлагаш. Само че това бих го направил ако имаше технически аргумент че без този горен протокол няма как да работи. Примерно някой биха сложили FTP или подобно да се транспортира върху mqtt. Други биха написали собствен и ще почнат да борят проблеми които са от десетилетия решени - в TCP, което реално отдолу пак си работи, ама няма да ползват функциите?
Само че за момента не съм видял такава лимитация в MQTT - TCP-то отдолу има достатъчно поддръжка за flow control, с ретрансмисии, ред на пакетите и тем подобни. Единственото е че MQTT не дава повече от 260мегаВата payload, ама хайде да го преживея засега. Не че е хубаво да има такива лимитации, но никой отсреща няма да ме съди за това.
А за заетостта - реално ти не комуникираш със сървъра, а с брокера, но е ясно какво имаш предвид. Начинът да кажеш на сървъра че си зает има няколко (на първо четене):
- по TCP - ползваш flow control-а за да забавиш идването
- по MQTT - като не можеш в момента да обслужиш нещо имаш UNSUBSCRIBE - отпиши се и никой няма да се мъчи да ти праща! И така е по-коректно, понеже освен да ти улеснява на тебе работата, MQTT има за идея да поддържа QoS. В тази ситуация, ако ти се отпишеш а друг е още закачен, ще стане така че автоматично ще има "failover".

gicho, няма нужда да ми обясняваш какво може и не може

Тоя код дето си дал пикае на всички възможни принципи. Разкарай го!
Този калбак е транспортното ниво на мрежата. Не може вътре да слагаш обработки от приложно ниво. Да ти обяснявам ли защо са измислени 7-те нива на OSI?
През този калбак минават всички сокети и почти всички протоколи, ти отде си сигурен, че само твоя mqtt клиент ще обработва всичко? Включая протоколите щото ако решиш да сложиш примерно секюрити - кво, ще пуснеш криптирания транфик към mqtt-то ли?
За многозадачност да не говорим...

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

- по TCP - ползваш flow control-а за да забавиш идването - това е достъпно ако стека е в твоя процесор.
- UNSUBSCRIBE. Така ще загубиш съобщения, не е решение.

Предполагам че се сещаш че за да напиша собствена библиотека съм навлизал сериозно в подробности . На твое място бих се замислил върху препоръката ми Няма проблем да разбиеш един файл на няколко номерирани съобщения.
Всички протоколи за трансфер на по голям обем данни имат flow control. MQTT не. Само това трябва да ти говори достатъчно. Не ми се навлиза в подробности повече.

И да имаш пред вид че в реални условия не получаваш винаги пакетите в реда който са изпратени

"- по TCP - ползваш flow control-а за да забавиш идването" при бавен нет ако преминеш таймаута за изпрашане на съобщението на брокера, ще те отсвири. При някои брокери е твърдо време. Така че и при теб да е стека, все тая, не можеш да разчиташ много на това.

Едно съобщение в повече да кажем NOACK щеше да е перфектно.

Деба, пак нещо брозера ми затри писанията... Може да е за добро. Сега по-синтезирано и без лирични отклонения:

Кажи сега, кой основополагащ принцип е нарушен с използването на функция/колбек, и нека не включваме субективни "принципи", известни още като "навици" в смисъла на "аз така го правя вече 20 години значи няма по-добро".
Според теб, LWIP-то "пикае" ли на тия споменатите принципи, или направо се*е? Има ли нещо ценно в него, или и него да хвърлям овреме? Да взема да пренапиша някой мегабайт сорс че ползвам функции и колбеци, да ги сменя (докато не са ме хванали) с polling на променливи в кооперативен суперлуп? Въх, там пък как да го направя без функции? А, да, ще си шатна всито код в меин-а и готово!
Функцията, както и варианта колбек, е нещо базово в нашите езици. Няма API дето да мине без такова - пък било то позикс, мозикс, или друго. А, има, някой хора "шерват" променливи, май тук имаше такава тема за борланд или ц++.
Не знам дали можеш да си представиш система която цъка само на прекъсвания - вЕрвай ми, има такива и са доста "responsive". Но това е залитане в друга посока и зависи от баланса между сложност на задачата и качество на хардуера на който ще тича. Да не го мислим - не отричам нуждата от тредове в определени ситуации и за определени цели.
Дай алтернатива - като код или като концепция, да видя дали ще мога да я смеля и как ще влезе в моя контекст?

Без да вземам отношение към MQTT, но това което Никола пише за редът на пакетите е абсолютно така. И още нещо, голям пакет при кофти връзка може никога да не мине, ама никога, толкова никога че ако изключиш кабела ще имаш по-голям шанс.

Разбира се че ще удари таймаута - то затова е сложен там. Т.е. определено не е за случаите когато се ребуутваш и ще те няма 1 минута - там трябва чисто да се откачиш от брокера, а той ще се погрижи да ти махне абонаментите, или не според clean session после. Чистото отписване си е нужно и при TCP, така че това е use case в който нещата са пределно ясни.
За случаите когато си натоварен има няколко сценария
- заявката можеш да е приемем (като payload) и си правиш ACK към брокера, но оттам данните при теб могат да почакат докато се освободиш и ги използваш - той брокера не се интересува от това
- ако нямаш място изобщо и за грам байт от заявката блокираш tcp-то или се отписваш, или се дисконектваш - както казахме, освен да мислиш за себе си е важно да дадеш информация навън че мислиш да подремнеш малко
Ако има NOACK какво да направи брокера с тази информация? Как да постъпи?
- да буферира заявката и да опита по-късно да ти я прати - докато му дадеш ACK? След колко време да препрати? Кога да се откаже (таймаут)? Не ти ли напомня на определени функции от tcp и защо да ги дублираме?
- да те прескочи от пращането - все едно да те брои за отписан (unsubscribe) но само за този пакет? това звучи смислено, но той вече е пуснал за търкаляне по транспорта пакета, ти ще трябва да го изчакаш целия и да го NAK-неш накрая (това май и сега може да стане откъм страната на клиента - при QoS 1 и 2 затварянето накрая може да се "обърка" умишлено)

Аз бих го погледнал така - из разните услуги които ползваме, физически или в интернет, дали ни допада варианта да пуснем един ден отпуска, да се натъкмим, въоръжим с папка документи, да минем 2 часа в трафика, да отидем в някоя държавна служба, да изчакаме 4 часа опашка и 1 час обедна почивка, да ни дойде реда, да извадим всички папки, да обясним целия казус и лелката отсреща да каже "NAK! Ела утре пак!"?
Сега излиза версия 5 на протокола и още не съм гледал какво ново има - може да са помислили и намерили такъв use case.