MQTT2_SERVER und QoS

[rudolfkoenig]

Dann drehe ich das Netzwerk wieder auf, alles geht auf "opened" aber das zuvor gepublishte kommt nicht bei dem MQTT2_SERVER der 2.FHEM Installation an.

MQTT2_CLIENT implementiert keine Warteschlange, und soweit ich weiss, auch andere MQTT-Clients machen das nicht. Mir ist kein Mechanismus bekannt, der fuer FHEM diese Funktionalitaet bereitstellt. Es waere mAn eine sinnvolle Erweiterung, weiss aber nicht wann/ob ich dazu komme.

[psycho160]

Ok danke für die Info mal, dann brauch ich nicht weiter zu debuggen. War mir eben wegen dem QoS nicht sicher ob ich was falsch mache.

Dachte QoS 1 stellt sicher das meine Nachricht ankommt, darum war ich verwirrt. Wäre aber auf jeden Fall toll wenn es sowas mal geben würde. Bis dahin kann ich nur hoffen, dass nicht ein Einbrecher genau in dem Moment kommt, wo der Link weg ist 

[rudolfkoenig]

Ich habe die Spec nochmal gelesen, und da steht:

When a Client reconnects with CleanSession set to 0, both the Client and Server MUST re-send any unacknowledged PUBLISH Packets (where QoS > 0)

insofern ist das Fehlen der Warteschlage bei QoS 1 ein Fehler. Es richtig zu loesen macht die Sache kompliziert, weil alle Messages persistent gespeichert werden muessen, was auch auf die Performance auswirkt. Vlt kann man mit einer unvolkommenen Loesung anfangen.

[psycho160]

Es richtig zu loesen macht die Sache kompliziert, weil alle Messages persistent gespeichert werden muessen, was auch auf die Performance auswirkt.

Kommt auf die Anzahl der generierten Messages an..

Vlt kann man mit einer unvolkommenen Loesung anfangen.

Vl. ja per attr dem Client sagen wieviele Messages er maximal speichern soll, die könnten ja in den Internals abgelegt werden (ja ich weiß das ist nicht 100% persistent aber mal ein anfang)
eventuell auch auf topic basis, das er nur die "wichtigen" speichert

[rudolfkoenig]

Ich habe jetzt eine erste Version von QoS=1 in MQTT2_CLIENT implementiert:

qosMaxQueueLength <number>
if set to a nonzero value, messages are published with QoS=1, and are kept in a memory-only buffer until acknowledged by the server. If there is no connection to the server, up to <number> messages are queued, and resent when the connection is esablished.

[psycho160]

CommandRef hört sich schon mal vielversprechend an. werds gleich testen.

[Beta-User]

@Rudi:Habe mir die Änderung mal angesehen, das ist ja im Prinzip sowas ähnliches wie die mit ACK versendeten Messages bei MySensors.

Mein Verständnis von Zeile 438ff ist, dass da einfach jede neue Message angehängt wird, bis eben die Queue voll ist, (erst) der überschießende Rest wird verworfen.

Da haben wir m.E. gleich zwei Probleme:
1. Eigentlich sollte eine neue Info die alte überschreiben, wenn sie auf denselben Topic geht, oder? (Vergleichbares macht https://svn.fhem.de/trac/browser/trunk/fhem/FHEM/00_MYSENSORS.pm#L846). (OT: Das ist übrigens ein gutes Argument, um das "one-json-for-all-über einen einzigen Topic"-Design mancher Clients zu kritisieren, das mir vom Bauchgefühl her schon immer irgendwie suspekt war...)

2. Dem Gefühl nach sollten von den so vorgefilterten, nicht überschriebenen Messages immer die ältesten gelöscht werden und nicht die jüngsten verworfen, das ganze also eine Art rollierendes System sein. Oder gibt es da eine irgendwie geartete Vorgabe?

(Sorry, ist schon klar, dass das zusätzliche Last ist, v.a., wenn jemand große Daten an viele Topics zu verwalten hat und meint, eine lange Queue sei super...).

[psycho160]

Erster Test erfolgreich, damit ist vorerst mal die Zustellung bei mir sichergestellt.

.... Eigentlich sollte eine neue Info die alte überschreiben, wenn sie auf denselben Topic geht, oder?  ...

Stimmt, ist ein gutes Argument. Könnte aber auch sein, dass die Messages "dazwischen" auch relevant sind, wenn der Empfänger sie weiterverarbeitet.. Denke da z.B an eine Türe die aufgemacht wird -> Alarm auslösen sollte -> und wider zugemacht wird. Dann kommt beim Server aber nur "zu" an, was wieder etwas verschleiern würde... Nur so ein Gedanke

[Beta-User]

Stimmt, ist ein gutes Argument. Könnte aber auch sein, dass die Messages "dazwischen" auch relevant sind, wenn der Empfänger sie weiterverarbeitet.. Denke da z.B an eine Türe die aufgemacht wird -> Alarm auslösen sollte -> und wider zugemacht wird. Dann kommt beim Server aber nur "zu" an, was wieder etwas verschleiern würde... Nur so ein Gedanke

An sich ist der Gedanke richtig, aber:

MQTT-mäßig meine ich, dass die Alarmierung bei abgebrochener Verbindung eine Sache ist, die mit dem Stichwort "LWT" zu lösen sein sollte. Denn andererseits hilft es ja nichts, wenn man ggf. erst Stunden oder Tage später von einem Problem erfährt, das gar nicht mehr relevant ist.

Aber da sind wir recht schnell bei der Frage, für was MQTT als Protokoll Sinn macht und für was nicht... MMn. ist es eher zur leichtgewichtigen Datenübertragung und langfristigen Überwachung geeignet, und eher weniger im Rahmen einer Alarmanlage oä. Dies vor allem wegen der prinzipbedingten Lücken, die sich spätestens dann ergeben, wenn der Client (FHEM oder ein anderer) oder eventuell auch nur der Server (Broker) neu gestartet wird. Und bei Tasmota würde ich z.B. auch annehmen, dass der neue Status den alten überschreibt (sofern das Thema da derzeit überhaupt abgehandelt ist...?).

[psycho160]

MQTT-mäßig meine ich, dass die Alarmierung bei abgebrochener Verbindung eine Sache ist, die mit dem Stichwort "LWT" zu lösen sein sollte. Denn andererseits hilft es ja nichts, wenn man ggf. erst Stunden oder Tage später von einem Problem erfährt, das gar nicht mehr relevant ist.

Mit LWT wird aber nicht sichergestellt, das eine Message "ankommt". Darum gibt es ja QoS und QoS=1 besagt, ich will das meine Nachricht beim Broker ankommt, egal was passiert und wieviele Tage es dauert.

In meinem Fall ist es auch nicht direkt eine Alarmanalge, sondern nur das Status-Display. Aber um das geht es auch gar nicht, sondern die Implementierung von Rudi oder anderen sollte sich an die Specs von MQTT halten.

Für mich ist ja das was mit dem aktuellen Update jetzt dazu gekommen ist schon super und deckt nun (zumindest meine) Anforderung ab und wäre schön wenn auch andere noch davon profitieren.

[Beta-User]

Hmm, habe zwar auf die Schnelle nicht die Spec gefunden, sondern "nur" https://www.hivemq.com/blog/mqtt-essentials-part-7-persistent-session-queuing-messages/, aber nach dem hast du recht, dass auch die "Zwischenmessages" zu queuen sein sollten:

The client must get all messages from a certain topic, even if it is offline. You want the broker to queue the messages for the client and deliver them as soon as the client is back online.

Es ist ausdrücklich von einem "certain topic" die Rede.

Daher stellt sich erst bei Resourcenlimitierung die Frage, wie gedroppt werden sollte, was dann ggf. ein komplexeres Thema wird (hier meine ich, dass neuer älter verdrängen sollte und dann bei "Ausgehen der topics" neuere ältere Topics verdrängen müßten, kann aber sein, dass es dazu auch abweichende Doku gibt...)

Was LWT angeht, ist klar, dass das eine etwas andere Frage ist, das sagt nur "Huston, wir haben..." und kann/soll dazu dienen, Schwachstellen (mehr oder weniger zeitnah) zu identifizieren.

[psycho160]

Daher stellt sich erst bei Resourcenlimitierung die Frage, wie gedroppt werden sollte, was dann ggf. ein komplexeres Thema wird (hier meine ich, dass neuer älter verdrängen sollte und dann bei "Ausgehen der topics" neuere ältere Topics verdrängen müßten, kann aber sein, dass es dazu auch abweichende Doku gibt...)

Ich hab es jetzt noch nicht genauer recherchiert in meinem System, aber ab wann sprechen wir denn von einer "Resourcenlimitierung"? Am Beispiel von einem Pi3 oder 4 denke ich müssen da schon einige 100 oder 1000 Messages in der Queue sein, dass sich hier performancemäßig etwas bemerkbar macht... Oder liege ich da falsch?

Ist vermutlich noch nicht real life getestet(tm) 

Nur so ne Idee:

Code Auswählen Erweitern

qosMaxQueueLength <topic1>:<number1> <topic2>:<number2>

Alles was nicht explizit in "qosMaxQueueLength" definiert wird als QoS=0 behandeln... Falls man ein MQTT Device hat was "spammt" kann man somit einen Überlauf abfangen

[Beta-User]

Na ja, wenn ich das heutige Attribut richtig verstanden habe, legt der user die Limitierung fest. Liegt die bei 10 oder 100 messages, kann es mit der Zeit schon eng werden, ohne dass der Server an sich irgendwie ins Schwitzen kommt. Auch  10000 Messages sind uU. schnell erreicht, da manche via MQTT_GENERAL_BRDIGE schlicht alle Events an einen externen Broker pusten. Hat man "gesprächige" Devices wie (z.B.) den ebus, Shelly@MQTT oder manche Wetterdienste, kann das schnell erreicht sein. Und dann gibt es noch "Spezialfälle", bei denen Bilddaten oder Logfiles via MQTT übertragen werden, bei denen größere Datenmengen zusammenkommen können (very un-mqtt-like imo, aber es wird eben gemacht oder zumindest angedacht.)

Ob die QoS über das  IO einstellen können sollte? Klingt für mich sehr kompliziert, und das ganze System für relativ wenige Anwendungsfälle mit vielen Prüfungen zu belasten, erscheint mir nicht optimal (obwohl es evtl. die "perfektere" Lösung wäre).
Der Weg, das über das einzelne Device einzustellen ist mit da irgendwie sympatischer; ist aber nur mal wieder mein "Bauchgefühl".

@Rudi:
Da wir es grade von MQTT_GENERAL_BRDIGE hatten: das erlaubt (auch) das Setzen der QoS-Flags. Bevor ich jetzt lange Code-Analysen anstelle die Frage, ob das von CLIENT irgendwie beachtet wird? (Wenn nein, sollte man dort ggf. die Doku anpassen; ich bin mir aber nicht sicher, ob 00_MQTT.pm sich da irgendwie anders verhält und das beachtet, _ glaube_ es aber eher nicht... Jedenfalls habe ich auf die Schnelle mit dem Suchwort qos via Trac nichts gefunden, was wie eine Queue@MySensors aussähe, das Ding geht also auch einfach von einer bestehenden Verbindung aus und schiebt es wohin (?), wenn diese nicht besteht...)
Ansonsten: in MQTT2_DEVICE ist mir zum Thema QoS bisher nichts aufgefallen, aber tendenziell würde ich annehmen, dass wenn, dann das Setzen des Flags eigentlich am besten bei der jeweiligen publish-Anweisung aufgehoben wäre (ähnlich :r)?

(Bin mir aber insgesamt unsicher; das ganze ist mal wieder sehr komplex, es wird vermutlich eine Unmenge Fragen dazu geben (je feiner man das einstellen kann, desto mehr) und  kaum jemand wird es wirklich brauchen... Aber wie hatte es neulich "jemand" formuliert: Man kann nie wissen, was am Ende rauskommt; plötzlich hat man doch ein weiteres IO-Modul für einen Client-TYPE usw, usf.. Daher wollte ich einfach meine Gedanken dazu mal aufschreiben.)

[rudolfkoenig]

Eigentlich sollte eine neue Info die alte überschreiben, wenn sie auf denselben Topic geht, oder?

Das ist mAn anwendungsspezifisch. Ich will die Sache nicht verkomplizieren, bevor ich konkrete Hinweise darauf habe, dass es noetig ist. Wer das Problem nicht mit qosMaxQueueLength 10000 loesen kann (entspricht ca 4MB), der moege sich mit Details melden.


MQTT_GENERAL_BRIDGE kann (genauso wie MQTT2_DEVICE) nur das retain Flag weitergeben.
In MQTT2_SERVICE habe ich nicht vor QoS=1 zu implementieren, da ich damit fuer alle unsauber beendete Clients eine Schlange speichern muesste. Ich behaupte, dass QoS=1 nicht fuer beliebig langes Zwischenspeichern bei Verbindungsproblemen gedacht ist, sondern zum Vermeiden von Paketverlusten bei UDP.

[psycho160]

Hallo,

seit der letzten Änderung (wo QoS 1 von Rudi implementiert wurde) scheint sich mein MQTT2_Client nach einiger Zeit "aufzuhängen". (siehe Log)

Mein Setup läuft einige Zeit, aber nach ca. paar Stunden sendet mein MQTT2_Client nichts mehr. Im Log steht discarding DISCONNECT (224)(0). HAt das was damit zu tun?

Man sieht auch im Log, dass meine Verbindung sehr instabil ist, aber das sollte doch für so geringe Datenmengen nicht relevant sein. Vermutlich mag der Client keine "unsauber" getrennten Verbindungen? Wenn ich FHEM neu starte läuft wieder alles.

Code Auswählen Erweitern

2020.07.02 06:19:21 1: 192.168.1.3:1883 reappeared (and_mqtt_client)
2020.07.02 06:20:11 1: 192.168.1.3:1883 disconnected, waiting to reappear (and_mqtt_client)
2020.07.02 06:20:14 1: 192.168.1.3:1883 reappeared (and_mqtt_client)
2020.07.02 06:21:02 1: 192.168.1.3:1883 disconnected, waiting to reappear (and_mqtt_client)
2020.07.02 06:21:03 1: 192.168.1.3:1883 reappeared (and_mqtt_client)


2020.07.02 06:21:52 1: 192.168.1.3:1883 disconnected, waiting to reappear (and_mqtt_client)
2020.07.02 06:21:53 1: 192.168.1.3:1883 reappeared (and_mqtt_client)
2020.07.02 06:22:43 1: 192.168.1.3:1883 disconnected, waiting to reappear (and_mqtt_client)
2020.07.02 06:22:43 5: and_mqtt_client: discarding DISCONNECT (224)(0)
2020.07.02 06:22:43 5: HttpUtils url=http://192.168.1.3:1883/
2020.07.02 06:22:43 4: IP: 192.168.1.3 -> 192.168.1.3
2020.07.02 06:22:46 5: and_mqtt_client: sending CONNECT (16)(29)(0)(6)MQIsdp(3)(2)(0)(30)(0)(15)and_mqtt_client
2020.07.02 06:22:46 5: SW: 101d00064d51497364700302001e000f616e645f6d7174745f636c69656e74
2020.07.02 06:22:46 1: 192.168.1.3:1883 reappeared (and_mqtt_client)
2020.07.02 06:23:15 3: MQTT2_DEVICE set mqtt_status_monitor and_water_on
2020.07.02 06:23:44 1: 192.168.1.3:1883 disconnected, waiting to reappear (and_mqtt_client)
2020.07.02 06:23:44 5: and_mqtt_client: discarding DISCONNECT (224)(0)
2020.07.02 06:23:44 5: HttpUtils url=http://192.168.1.3:1883/
2020.07.02 06:23:44 4: IP: 192.168.1.3 -> 192.168.1.3
2020.07.02 06:23:45 5: and_mqtt_client: sending CONNECT (16)(29)(0)(6)MQIsdp(3)(2)(0)(30)(0)(15)and_mqtt_client
2020.07.02 06:23:45 5: SW: 101d00064d51497364700302001e000f616e645f6d7174745f636c69656e74
2020.07.02 06:23:45 1: 192.168.1.3:1883 reappeared (and_mqtt_client)
2020.07.02 06:24:32 3: MQTT2_DEVICE set mqtt_status_monitor and_haus1_zu
2020.07.02 06:24:37 1: 192.168.1.3:1883 disconnected, waiting to reappear (and_mqtt_client)
2020.07.02 06:24:37 5: and_mqtt_client: discarding DISCONNECT (224)(0)
2020.07.02 06:24:37 5: HttpUtils url=http://192.168.1.3:1883/
2020.07.02 06:24:37 4: IP: 192.168.1.3 -> 192.168.1.3
2020.07.02 06:24:46 5: HttpUtils url=http://192.168.1.3:1883/
2020.07.02 06:24:46 4: IP: 192.168.1.3 -> 192.168.1.3
2020.07.02 06:24:46 5: and_mqtt_client: sending CONNECT (16)(29)(0)(6)MQIsdp(3)(2)(0)(30)(0)(15)and_mqtt_client
2020.07.02 06:24:46 5: SW: 101d00064d51497364700302001e000f616e645f6d7174745f636c69656e74
2020.07.02 06:24:46 1: 192.168.1.3:1883 reappeared (and_mqtt_client)
2020.07.02 06:25:35 1: 192.168.1.3:1883 disconnected, waiting to reappear (and_mqtt_client)
2020.07.02 06:25:35 5: and_mqtt_client: discarding DISCONNECT (224)(0)
2020.07.02 06:25:35 5: HttpUtils url=http://192.168.1.3:1883/
2020.07.02 06:25:35 4: IP: 192.168.1.3 -> 192.168.1.3
2020.07.02 06:25:37 5: and_mqtt_client: sending CONNECT (16)(29)(0)(6)MQIsdp(3)(2)(0)(30)(0)(15)and_mqtt_client
2020.07.02 06:25:37 5: SW: 101d00064d51497364700302001e000f616e645f6d7174745f636c69656e74
2020.07.02 06:25:37 1: 192.168.1.3:1883 reappeared (and_mqtt_client)
2020.07.02 06:25:38 5: and_mqtt_client: received RESERVED_0 (2)(0)(0) (2)(0)(0) (2)(0)(0) (2)(0)(0) (2)(0)(0) (2)(0)(0) (2)(0)(0) (2)(0)(0)
2020.07.02 06:25:38 1: M2: Unhandled packet RESERVED_0, disconneting and_mqtt_client
2020.07.02 06:25:38 5: and_mqtt_client: discarding DISCONNECT (224)(0)
2020.07.02 06:25:38 1: 192.168.1.3:1883 disconnected, waiting to reappear (and_mqtt_client)
2020.07.02 06:25:38 5: and_mqtt_client: received RESERVED_0
2020.07.02 06:25:38 1: M2: Unhandled packet RESERVED_0, disconneting and_mqtt_client
2020.07.02 06:25:38 5: and_mqtt_client: discarding DISCONNECT (224)(0)
2020.07.02 06:25:38 5: HttpUtils url=http://192.168.1.3:1883/
2020.07.02 06:25:38 4: IP: 192.168.1.3 -> 192.168.1.3
2020.07.02 06:25:40 5: and_mqtt_client: sending CONNECT (16)(29)(0)(6)MQIsdp(3)(2)(0)(30)(0)(15)and_mqtt_client
2020.07.02 06:25:40 5: SW: 101d00064d51497364700302001e000f616e645f6d7174745f636c69656e74
2020.07.02 06:25:40 1: 192.168.1.3:1883 reappeared (and_mqtt_client)

List meines IODevices:

Code Auswählen Erweitern

Internals:
   BUF          
   DEF        192.168.1.3:1883
   DeviceName 192.168.1.3:1883
   FD         16
   FUUID      5ef9ab6c-f33f-f34f-8edf-3f2fc62221cd4c73
   NAME       and_mqtt_client
   NR         70
   PARTIAL   
   STATE      opened
   TYPE       MQTT2_CLIENT
   WBCallback
   clientId   and_mqtt_client
   connecting 2
   lastMsgTime 1593663938.25785
   nextOpenDelay 5
   qosCnt     12
   qosMaxQueueLength 10
   READINGS:
     2020-06-30 11:52:31   lastPublish     fhem/status/light/waterpump:waterpump
     2020-07-02 06:26:37   state           opened
   qosQueue:
Attributes:
   qosMaxQueueLength 10
   room       xSystem->MQTT
   verbose    5