Replies: 12 comments 28 replies
-
|
@robertsLando
Which carriers do you want to support ?
I see quite some config options to tune the stack, what options do you want to keep? Do you want to support MQTTv5? (I personally think its a bad move from architectural perspective as part of the application layer seems to be incorporated in the transport layer ,e.g. properties) Then some other notes:
I'm happy to contribute code from Opifex if that helps.
Hope this helps! Kind regards, |
Beta Was this translation helpful? Give feedback.
-
Yeah I mean better bundling
Supporting webstreams as well would be nice
I think we should somehow make those transports threeshakable, I mean letting people to use them but place them outside the scope of mqtt, like a pre-step where use needs to register the transport plugin like
Actually we already support MQTTv5 so yes it should still support v5
I place both at the same level. The important thing, IMO, is to find an acceptable trade-off between the two. Currently, MQTTjs is the most used NodeJS client, so I don't want to make it less performant, but keeping actual performances or better than the actual ones is required.
If we drop mqtt-packet we would need to firstly ensure all tests of the actual mqtt packet are passed by yours. Also it's an important part of the mqtt client so also a benchmark would be required |
Beta Was this translation helpful? Give feedback.
-
|
OK, then I guess Opifex will get MQTTv5 support as well, at least client side, and maybe also browser support if I can do that without external dependencies 😄 What do you suggest we do next? Kind regards, |
Beta Was this translation helpful? Give feedback.
-
|
Ok, I did some benchmarking on the generation of packets using: https://github.com/mqttjs/mqtt-packet/tree/master/benchmarks/generate.js The results were quite variable on my codespace so I jacked up the max to 10.000.000 The results were as follows: mqtt-packetTotal time 7213 opifexTotal time 12362 1.3M packets per second vs 0.8M packets per second, so mqtt-packet is about 50% faster than Opifex. If you have any ideas then I'm curious to learn! The source is in: And as generate.js for Opifex I used: const mqtt = require('@seriousme/opifex/mqttPacket')
const max = 10000000
let i
const buf = Buffer.from('test')
const PUBLISH = 3
// initialize it
mqtt.encode({
type: PUBLISH,
topic: 'test',
payload: buf
})
const start = Date.now()
for (i = 0; i < max; i++) {
mqtt.encode({
type: PUBLISH,
topic: 'test',
payload: buf
})
}
const time = Date.now() - start
console.log('Total time', time)
console.log('Total packets', max)
console.log('Packet/s', max / time * 1000)I will have a look at parsing performance as well as I'm curious how that plays out! Kind regards, |
Beta Was this translation helpful? Give feedback.
-
|
Super interesting benchmarks :) Did you also tried with decoding as well? About the performance impact I think that 50% is a bit too much of a loss, would be worth investigating why this happens. IMO everything is due to @seriousme did you tried to run the same benchmark on browser? IMO what we could consider is: |
Beta Was this translation helpful? Give feedback.
-
Haven't tried it on a browser, but the raw encode/decode (without sockets etc) does not contain anything around timing nor (apart from decode on mqtt-packet) nor streams. The opifex codec only works in memory. Its Opifex' MQTTCon that communicates with the socket (via a ReadableByteStream) and uses a minimum number of reads (max 5 per packet: first byte, each packetsize byte and once the size is known the remaining size) and only a single write. Anyways: feel free to test any way you like, and either stick with the current streaming parser or switch to a blockbased parser like Opifex uses. I have no horse in this race, just trying to explain how I did it. Hope this helps. Kind regards, |
Beta Was this translation helpful? Give feedback.
-
|
I have no preference neighter, I just want to be sure to take the best choice by evaluating all pro and cons of each one :)
That's nice! Feel free to tag me in your PRs
Could you explain a bit more? Because there are many different ways to achieve this |
Beta Was this translation helpful? Give feedback.
-
|
I'll tag you (if I don't forget it 😉 ) . Opifex uses separate files per packetType, ech containing both the encoder and the decoder for that packet. My current plan (not carved in stone), is to add Set of allowable properties to each packet file and verify those on decoding and encoding. This way I can reject them if they are not defined for that packet. Something along the lines of: const associatedProps = new Set([Prop1,Prop7,Prop9])And then in encode decode This might be optimizable by using bitmap,but the property ID's go up to 42 and I'm not sure if a javascript number will be able to deal with so many bits without resorting to bigInt or manual splitting it across 2 numbers. It might be more effecttive to code the bitmap the other way around: making an array of 42 bytes and coding the packets that are allowable Hope this inspires! Hans |
Beta Was this translation helpful? Give feedback.
-
|
Interestingly parsing gives a different picture. mqtt-packetTotal packets 10000000 OpifexTotal packets 10000000 Here mqtt-packet does about 0.7M packets per second where Opifex is doing 2.1M packets per second, about 3 times as much. The code I used for Opifex was: const mqtt = require('@seriousme/opifex/mqttPacket')
const max = 10000000
let i
const start = Date.now() / 1000
for (i = 0; i < max; i++) {
mqtt.decode(Buffer.from([
48, 10, // Header (publish)
0, 4, // Topic length
116, 101, 115, 116, // Topic (test)
116, 101, 115, 116 // Payload (test)
]))
}
const time = Date.now() / 1000 - start
console.log('Total packets', max)
console.log('Total time', Math.round(time * 100) / 100)
console.log('Packet/s', max / time)Btw: this is NodeJS 24.1, the previous generate test was NodeJS 22.16 mqtt-packet generate.jsTotal time 9523 Opifex generate.jsTotal time 12806 And somehow mqtt-packet got a bit slower and Opifex got slightly faster 🤔 KInd regards, |
Beta Was this translation helpful? Give feedback.
-
|
Here the problem is that streams support backpressure and while opifex could look better in performances it would crash on a long running mqtt client. When there is a high load the backpressure kicks in and slows down parsing, you could notice this on a long running task, you would see that opifex parser dies at some point |
Beta Was this translation helpful? Give feedback.
-
|
This is the reason why |
Beta Was this translation helpful? Give feedback.
-
The backpressure is automatically handled by the onderlying stream, ie. the socket will be "paused" if the stream isn't processing fast enough. I haven't tested it but on paper this should not be a problem. Mind you I only tested the performance of the raw decoder, not the decoder on top of the socket stream. When I find some time I will see if I can test that as well, but feel free to try to test it yourself. The ideal solution would af course be a BYOB that directly gets the data from the C++ socket like Deno does, as that saves another copy of data. But as I mentioned before, I doubt whether all this performance optimization translates into real world benefits. 1M packets per second on a single core should cover most usecases I think and for those that need more , the costs of extra cores is typically not a huge bottleneck. Kind regards, |
Beta Was this translation helpful? Give feedback.
-
|
Btw: I did some analysis on the generate.js using mqtt-packet[Summary]:
opifex[Summary]:
With Opifex spending most of its C++ on:
|
Beta Was this translation helpful? Give feedback.
-
|
Ok, I just released Opifex 1.9.4 which replaces: return Uint8Array.from([
(packetType << 4) | flags,
...encodeLength(bytes.length),
...bytes,
]);by: const encodedLength = encodeLength(bytes.length);
const buflenght = 1 + encodedLength.length + bytes.length;
const buffer = new Uint8Array(buflenght);
let offset = 0;
buffer[offset++] = (packetType << 4) | flags;
for (let i = 0; i < encodedLength.length; i++) {
buffer[offset++] = encodedLength[i];
}
for (let i = 0; i < bytes.length; i++) {
buffer[offset++] = bytes[i];
}
return buffer;
}Which avoids a number of copy rounds as there is no destructure anymore and Uint8Array.from also does some cyping internally apparently. I need to run another test on codespace, but a local test was already more performant. Kind regards, |
Beta Was this translation helpful? Give feedback.
-
|
Not yet, I tried it last weekend but then the performance of my codespace was all over the place both for mqtt-packet and for opifex. Tests took between 7 and 14 seconds, I presume my codespace was suffering from a noisy neighbor. |
Beta Was this translation helpful? Give feedback.
-
|
Just made a script that ran generate.js 10x mqtt-packet and then opifex and put the results per round in 1 row. Had to drop 5 of the resulting rows because either mqtt-packet or opifex was on about 1/10 of the performance in the other rows. Then took the avarage of the remaining rows and that resulted in 125% performance of mqtt-packet over opifex. My expectation was that if I skip the regexes the test for valid UTF8 and valid topic structure the performance will be more or less equal, so I tried that as well. Not in the original Opifex, but in the version installed in my codespace. If you have any ideas then please shout ! |
Beta Was this translation helpful? Give feedback.
-
|
also tried replacing the |
Beta Was this translation helpful? Give feedback.
-
|
@seriousme did you checked |
Beta Was this translation helpful? Give feedback.
-
|
I have seen |
Beta Was this translation helpful? Give feedback.
-
|
Ok, I also noticed that the Opifex encoder performs more checks. If you really want top notch performance then the most optimal encoder I can think of does the following:
You could even keep record on what percentage of encodings does buffer.concat() and use that to tune X. One step even further would be to allow the user to generate a packet , given topic, and payloadsize and then generate then packet once and offer a method which only overwrites the payload and maybe the packet number (for QoS > 0) For Opifex I like the current way of checking everything for validity, but I can imagine you might want to make other choices. Hope this helps, |
Beta Was this translation helpful? Give feedback.
-
|
The format is nearly the same but Opifex uses UInt8Array where mqtt-packet uses buffer and some property names are slightly different. The Opifex parser is 3x as fast (yes, that is 300% 😉) as the MQTT packet parser when given a Buffer. I'll see what I can do to find out what is hindering the generator. My guess is that we are missing out on a V8 optimzation. Not that I think there is practical value in squeezing out the last CPU cycles, I'm just curious. |
Beta Was this translation helpful? Give feedback.
-
|
Did some more research and it starts to feel like a black art now. I created the fastest MQTT publish packet (pubPacket) encoder that I could think of and still encodes every MQTT v3 packet you throw at it, unless its size is bigger than 1Kb. const utf8Encoder = new TextEncoder();
const BitMask = {
bit0: 2 ** 0,
bit1: 2 ** 1,
bit2: 2 ** 2,
bit3: 2 ** 3,
bit4: 2 ** 4,
bit5: 2 ** 5,
bit6: 2 ** 6,
bit7: 2 ** 7,
};
const PacketType = { 'publish': 3 }
class mqtt {
// {
// cmd: 'publish',
// topic: 'test',
// payload: buf
// }
generate(packet) {
const packetType = PacketType[packet.cmd];
if (packetType === undefined) {
throw new Error(`Unknown packet type: ${packet.cmd}`);
}
const topic = utf8Encoder.encode(packet.topic)
const topicLength = topic.length;
const payload = packet.payload;
const payloadLength = payload.length;
const messageId = packet.messageId || 0; // Default to 0 if not provided
const qos = packet.qos || 0;
const flags = (packet.dup ? BitMask.bit3 : 0) +
(qos & 2 ? BitMask.bit2 : 0) +
(qos & 1 ? BitMask.bit1 : 0) +
(packet.retain ? BitMask.bit0 : 0);
const qosLength = qos > 0 ? 2 : 0; // QoS level requires 2 bytes if > 0
const remainingLength = qosLength + 2 + topicLength + payloadLength; // 2 bytes for topic length
const result = new Uint8Array(1024); // Placeholder for the result buffer
let offset = 0;
// Write the fixed header
result[offset++] = (packetType << 4) | flags;
// Write the remaining length
let remainingLengthEncoded = remainingLength;
do {
let byte = remainingLengthEncoded % 128;
remainingLengthEncoded = Math.floor(remainingLengthEncoded / 128);
if (remainingLengthEncoded > 0) {
byte |= 128; // Set the continuation bit
}
result[offset++] = byte;
} while (remainingLengthEncoded > 0);
// Write the messageId if applicable
if (qosLength !== 0) {
result[offset++] = (messageId >> 8) & 0xFF; // MSB
result[offset++] = messageId & 0xFF; // LSB
}
// Write the topic length
result[offset++] = (topicLength >> 8) & 0xFF; // MSB
result[offset++] = topicLength & 0xFF; // LSB
// Write the topic
result.set(topic, offset);
offset += topicLength;
// Write the payload
result.set(payload, offset)
offset += payloadLength;
// Return the result as a new Uint8Array with the correct length
return result.subarray(0, offset);
}
}
module.exports = new mqtt();It only creates a Uint8Array of size 1024bytes once and then directly stuffs the data in there, the result.subarray should be a view and not a copy. Only one alloc, should be pretty quick without all the overhead.
The superslim pubPacket is actually way more slow than both mqtt-packet and opifex! I do understand now why mqtt-packet is way quicker than opifex in benchmarks/generate.js. |
Beta Was this translation helpful? Give feedback.
-
Could you think about having the same format? I mean same properties name. It would make the transition easier :) |
Beta Was this translation helpful? Give feedback.
-
|
Having the exact same interface would break compatibility for Opifex users (about 1800 downloads a month according to NPM and then there is Deno's JSR.io) but more importantly it would lock both projects hard together and that might limit them both. If you like the Opifex packet parser/generator setup (once it supports v5 and I have slashed the last performance hickups in the generator) we could copy it into MQTTjs (Opifex is MIT licenced) and then both projects can do whatever they like. Once the parser/generator is stable I doubt whether it will receive many changes until v6 comes around (if ever). What do you think? Kind regards, |
Beta Was this translation helpful? Give feedback.
-
This is the key point of how it's structured. There is absolutely no need to flatten it. Note that bl got really slower over the years (it was faster) and we had to reimplement it recently for another project: https://github.com/platformatic/kafka/blob/main/src/protocol/dynamic-buffer.ts. We plan to ship it as a new module soon. |
Beta Was this translation helpful? Give feedback.
-
|
I checked a bit more and the mqtt-packet test does call concat to create the Buffer, so we were comparing apples to apples yesterday. So where does the difference come from? I set iterations per round to 1.000.000 (1/10 of yesterday) to speed things up a bit. I modified pubPacket to allocate just the amount of bytes required for the packet instead of 1K, that alreaday makes quite a difference: allocate exact packet size instead of 1K
Then I removed the .subArray() removed
And for good measure I also changed Buffer.allocUnsafe()
As you can see the numbers are still a bit variable across rounds. But I think that In general we can conclude that:
Hope this helps. |
Beta Was this translation helpful? Give feedback.
-
|
@seriousme thanks for all this research, it's super interesting and I think @mcollina could add a good context to this as he is a performance master and has a deep knowledge of NodeJS at lower level :) I think point 1 and 2 looks to be the ones that have the higher impact on performances, could you do something on opifex side with this information to improve it? |
Beta Was this translation helpful? Give feedback.
-
|
I have added a BL like structure to the Opifex encoder constructor(packetType:TPacketType) {
this.buffers = [];
this.numBytes = 0;
this.packetType = packetType;
}
addArray(value:number[] | Uint8Array){
this.buffers.push(value);
this.numBytes += value.length;
}
addByte(value:number){
this.buffers.push([value]);
this.numBytes++;
}encoder.done() now does the following: done(flags:number):Uint8Array {
const packetType = this.packetType;
const encodedLength = encodeLength(this.numBytes);
const totalLength = 1 + encodeLength.length + this.numBytes;
const result = new Uint8Array(totalLength);
let pos = 0
result.set([packetType << 4 | flags], pos++);
result.set(encodedLength, pos);
pos += encodedLength.length;
for (const buffer of this.buffers) {
result.set(buffer, pos);
pos += buffer.length;
}
return result;
}This increases its speed significantly but it doesn't bridge the whole gap yet.
I also tried calling the encoder directly , without going through the lookup array that matches packet identifier to find the encoder for the publish packet but that doesn't make much of a difference. Kind regards, |
Beta Was this translation helpful? Give feedback.
-
|
They say hindsight is 20/20. I think I found the difference: This also explains why the Buffer.allocUnsafe nor the removal of the rexExp did anything, apparently those items were dwarfed by the TextEncoder elephant in the room. Just to see how fast it could be, I changed pubPacket.js (the fastest implementation I could think of , listed above) and used:
pubPacket using Buffer.write()
Taking the average and setting PubPacket to 100 we get
Which seems to leave quite some headroom for mqtt-packet to improve performance 😉 Hope this helps, it's been an interesting journey! Kind regards, |
Beta Was this translation helpful? Give feedback.
-
|
Where is the code of this benchmark? Can you please assemble all this data so it's easier to unpack? All of this discussion is really hard to follow unfortunately (given the non-linearity of discussions). |
Beta Was this translation helpful? Give feedback.
-
|
Where is the code of this benchmark? Can you please assemble all this data so it's easier to unpack? All of this discussion is really hard to follow unfortunately (given the non-linearity of discussions). |
Beta Was this translation helpful? Give feedback.
-
|
@mcollina , @robertsLando It uses:
Hope this helps! Kind regards, |
Beta Was this translation helpful? Give feedback.
-
|
I think you are completely missing the point on the writeToStream concept and avoiding a massive amount of buffer allocations vs generating a full packet at any given time. |
Beta Was this translation helpful? Give feedback.
-
|
I just tried encodeInto() this morning and added it to generateAll() in https://github.com/seriousme/mqtt-packet/tree/benchmark-publish-packet The results were predictable, it is by far the slowest solution.
It seems like more people are complaining about TextEncoder/TextDecoder performance: whatwg/encoding#343 Kind regards, |
Beta Was this translation helpful? Give feedback.
-
|
I think that we both agree that for high speed data processing:
What I learned from the whole experience is that:
Kind regards, |
Beta Was this translation helpful? Give feedback.
-
|
I tried to summarize everything in: #2002 let's continue our discussion there |
Beta Was this translation helpful? Give feedback.
Uh oh!
There was an error while loading. Please reload this page.
-
@robertsLando
As discussed in moscajs/aedes-persistence#105 (comment) discuss how to improve.
Beta Was this translation helpful? Give feedback.
All reactions