uWebsockets publish dropping messages

[rpgehlot]

Hi @alexhultman ,
Thanks for this great and efficient library which i have to started to use recently in place of socket.io which was giving us horrible performance in production.
Basically we are using it in context of a trading platform which is in house to our firm. There we get around ~400 msg/sec of ~2KB with ~60 connected websocket clients. We are using uWebsockets pub-sub with maxBackPressure setting of 6KB.
Everything was fine, but we noticed some messages are being dropped. I understand its happening because of built up backpressure because of which it drops messages for slow client or either drops their connection.
Also i tried using ws.send method and manage backpressure using ws.getBufferedAmount() and implement my own pub-sub over it but obviously it was much slower and useless.

I really want to use your library becuase its way efficient than others, but if you can help me to fix this problem of message drop, that would be very grateful of you.

[hst-m]

I understand its happening because of built up backpressure because of which it drops messages

Are you saying you'd prefer it drop the connection rather than drop messages? Docs need to be updated, there is new option closeOnBackpressureLimit to drop connection rather than drop message. If you don't want it doing either you should raise limit

[ghost]

Hi, https://github.com/uNetworking/uWebSockets/releases/tag/v18.22.0 did introduce closeOnBackpressureLimit but it looks like that option was never wrapped here.

[ghost]

@rpgehlot Can you test the very latest binaries? https://github.com/uNetworking/uWebSockets.js/tree/binaries

they should have closeOnBackpressureLimit option

[rpgehlot]

Thanks for your prompt reply, but i needed centos 7 binary and i actually dont have root access so can't build that bin right now. will get it built and try out it this option.

[rpgehlot]

Hi, I tried your new binary it seems to work, but i have a question if i increase the maxBackpressure to 11MB for e.g, how would it affect the latency of packets reaching the browsers ? would the perf suffer because of high maxbackpressure ?

[rpgehlot]

I am seeing poor performance with maxBackpressure=11MB and closeOnBackpressureLimit : true and no drops, and low latency with maxBackpressure=6KB and closeOnBackpressureLimit:false but with message drops, i guess there is no middle ground. or is there ?

[ghost]

11MB backpressure sounds like a lot if you care about latency. If latency is a concern, set backpressure low. But either way you cannot just shove messages like there's no tomorrow - if you constantly end up with messages stuck in backpressure then you're sending too quick.

[rpgehlot]

So, if we can't control that incoming rate, would some kind of rate limiter would be helpful which can control outward messages sent per sec to each client ? and is there a possibility we can use such a thing with publish ?

[hst-m]

i guess there is no middle ground. or is there ?

11MB is 2000 times larger than 6KB, the middle ground would be all the values in between

we get around ~400 msg/sec of ~2KB with ~60 connected websocket clients

each message is 2KB? so you are publishing 800KB per second to 60 clients? or is that 2KB per sec, 5 bytes per message?

[rpgehlot]

thats the worst case figures i have mentioned, message rate is actually variable depending on the trading volume. So yes it would amount to 800KB per sec for a single client and we have 60 such clients. most of the messages are 2KB but initial first message amounts to a little bigger value upon connection start, so thats why i used 11MB. for case when a single message is > than maxbackpressure, if i use a smaller value of 6KB then it actually closes the connection at start itself and stuck in a infinite loop since i used closeOnBackpressureLimit : true

[uasan]

I think you need to pause before sending the next message if there is backpressure on the socket

if (socket.write(message) === false) {
  await socket.onDrain();
}

[hst-m]

So you tried 6KB which is smaller than a single message, and you tried 11MB which is 14 seconds of 800KB/sec. Have you considered trying anything between those 2 values?

800KB/sec seems like a lot, are you sure these are efficiently packed messages? Binary format etc?

[rpgehlot]

No i haven't tried anything b/w. I encode json messages to arraybuffer(binary message) as that is preferred in uWS (from docs).
Currenlty i am sending all messages using
uApp.publish(<message_room>, binarybuffer, true, true) and it should do compression as last flag is true.
I am thinking to break the bigger message in chunks size smaller than maxbackpressure, and send using ws.send if there is no backpressure else wait for drain to send the next chunk. So a combination of ws.send and ws.publish but publishing only when whole chunk is sent. (as there io no order guarantee b/w publish and send).
This hopefully should give me the perf boost using smaller maxbackpressure and also to avoid the problem of connection close on first message.

[uasan]

On client (Chrome), this problem is nicely solved with interface WebSocketStream
https://web.dev/websocketstream/

You need to do something like this for the server

[hst-m]

I encode json messages to arraybuffer

What does this mean? does not make any sense. JSON is a utf8 string of encoded values. Sounds like you are sending JSON and trying to compress it which is inefficient way to send large amounts of data

No i haven't tried anything between

You may want to try something between 6KB and 11MB if you are looking for middle ground

[rpgehlot]

Can you suggest a good soln for efficiently sending large data ?
Ok will try

[hst-m]

JSON vs binary:

// data to send
data={field1:3751923517,field2:1842963572}

// JSON encode
ws.send(JSON.stringify(data)) // 41 bytes of JSON utf8 string + cpu time to encode

// Binary encode
buf=Buffer.allocUnsafe(8)
buf.writeUInt32BE(data.field1)
buf.writeUInt32BE(data.field2,4)
ws.send(buf,true) // 8 bytes of Binary

JSON with 41 bytes is sending 5 times more than the 8 bytes of Binary and it uses more CPU to encode string. So your 800KB/sec would then be 160KB/sec. Probably would help your backpressure issue

[uasan]

When choosing a communication format, it is important not only the size and speed, but also portability to clients, if you do not pet project, then MessagePack is a reasonable

[kirbysayshi]

portability is important, absolutely. Another is library size, which msgpackr seems to be reasonable: https://bundlephobia.com/[email protected] Given that the msgpackr benchmark includes avsc, and the benchmark I linked had avsc at similar speeds to js-binary, then speed is not the comparison :) . bandwidth, portability, library size (client impact), API design, etc become the deciding factors.

[hst-m]

All these published benchmarks depend on the data used, I coded tool to compare different data types and encoding methods including avsc, msgpack, json, and custom binary:

benchmark.js (click to expand)

// encode data
const data={field1:3751923517,field2:1842963572}

const msgpack=require('@msgpack/msgpack')
const avscSchema=require('avsc').Type.forValue(data)

const encodeFns={
   binary:data=>{
      const buf=Buffer.allocUnsafe(8)
      buf.writeUInt32BE(data.field1)
      buf.writeUInt32BE(data.field2,4)
      return buf
   },
   avsc:data=>avscSchema.toBuffer(data),
   msgpack:data=>msgpack.encode(data),
   json:data=>Buffer.from(JSON.stringify(data)),
}

// node benchmark.js [encodings=all] [iterations=2_000_000]
const encodings=process.argv[2]&&process.argv[2]!='all'?[process.argv[2]]:Object.keys(encodeFns)
const iterations=process.argv[3]||2_000_000

encodings.forEach(type=>{
   let i=iterations,time=process.hrtime.bigint()
   while(i--)encodeFns[type](data)
   console.log(`${(type+':     ').slice(0,9)} encode time(ns): ${Math.floor(Number(process.hrtime.bigint()-time)/iterations)} - bytes: ${encodeFns[type](data).length}`)
})

Encoding	time (ns)	bytes
binary	60	8
avsc	200	9
msgpack	1350	25
json	850	41

[rpgehlot]

I ended up doing custom binary protocol for my app like in your first comment, to send a message and that turned messagesize from 2KB to under 200bytes which brought down 800kbps to 80 kbps. Using external libraries mentioned here was not giving much reduction in size because it keeps strings as strings, while i converted most strings into 8 and 16 bits integers using a mapping which gave this 90% reduction in size and greatly improved throughput. Thanks a lot!!

[hst-m]

Wow nice work that was fast, only couple days and you got it figured out. Sounds like exactly what you needed to do map strings to integers, strings are usually the slow point, no library would do that for you. 10 times reduction is not bad, thanks for sharing :)

[rpgehlot]

I have a question regarding publish, if a websocket is subscribed to two rooms( r1, r2) and if i publish message on r1 first
using uApp.publish(r1, buf1, true) and then on r2 same way, is the order guaranteed here or message published on r2 can reach client first ?

[ghost]

Order is only guaranteed inside a topic - your example is nondeterministic

[hst-m]

@rpgehlot Do you have fields using less than 1 byte of data? True/false only needs 1 bit, fields with 4 values need 2 bits, 8 values need 3 bits etc, but the minimum Buffer can write is 1 byte/8 bits, to save space on those you pack the byte with multiple fields, if your data has many small value fields it saves lots of space, I use this function to pack bytes:

const packBytes=(...bitsVals)=>bitsVals.reduce((byte,val,i)=>i%2?byte+val:byte<<val,0)>>>0

Tell it how many bits to use per field and the value of the field:

// example data
data={field1: true, field2: false, field3: 3, field4: 15} // fields use 1 bit, 1 bit, 2 bits, 4 bits

// packBytes(bits, value[, bits, value][, bits, value]..) values are positive integers
packed=packBytes(1, data.field1, 1, data.field2, 2, data.field3, 4, data.field4) // packed.toString(2) == '10111111'

buf.writeUInt8(packed)

Then you unpack it with function telling it the same bit sizes:

// unpackBytes(byte, bits[, bits][, bits]..)
const unpackBytes=(byte,...bits)=>{for(let i=bits.length-1;i>=0;i--){let val=byte%(1<<bits[i]);byte>>>=bits[i];bits[i]=val}return bits}

byte=dataView.getUint8()
[field1,field2,field3,field4]=unpackBytes(byte, 1, 1, 2, 4) // field1 == true; field2 == false; field3 == 3; field4 == 15

You can also use the same functions to pack data into 16 or 32 bit spaces if more room is needed, this allows you to be most efficient without wasting any bits. Maybe you have an array of items that need 32 bits each, then you pack it with exactly how many bits each field needs

Compare byte size with other encodings:

// encode data
const data={field1:true,field2:false,field3:3,field4:15}

const msgpack=require('@msgpack/msgpack')
const avscSchema=require('avsc').Type.forValue(data)
const packBytes=(...bitsVals)=>bitsVals.reduce((byte,val,i)=>i%2?byte+val:byte<<val,0)>>>0

const encodeFns={
   packBytes:data=>{
      const buf=Buffer.allocUnsafe(1)
      buf.writeUInt8(packBytes(1,data.field1,1,data.field2,2,data.field3,4,data.field4))
      return buf
   },
   avsc:data=>avscSchema.toBuffer(data),
   msgpack:data=>msgpack.encode(data),
   json:data=>Buffer.from(JSON.stringify(data)),
}

// node benchmark.js [encodings=all] [iterations=2_000_000]
const encodings=process.argv[2]&&process.argv[2]!='all'?[process.argv[2]]:Object.keys(encodeFns)
const iterations=process.argv[3]||2_000_000

encodings.forEach(type=>{
   let i=iterations,time=process.hrtime.bigint()
   while(i--)encodeFns[type](data)
   console.log(`${type} encode time(ns): ${Math.floor(Number(process.hrtime.bigint()-time)/iterations)} - bytes: ${encodeFns[type](data).length}`)
})

Encoding	time (ns)	bytes
packBytes	65	1
avsc	235	4
msgpack	1475	33
json	765	53

[rpgehlot]

Hi this seems like a great optimization to save more bytes, but i have a question

1. how will we store double/string values here and in this signature packBytes(bits, value[, bits, value][, bits, value]..) bits is the minimum number of bits required to represent a value ?
2. what if instead of this data={field1: true, field2: false, field3: 3, field4: 15} // fields use 1 bit, 1 bit, 2 bits, 4 bits, i had
   data={field1: true, field2: false, field3: 35.67, field4: 'tatamotors' } // fields use 1 bit, 1 bit, x bits, y bits. (x + y > 6)
   we can't pack this second data into 1 byte, so do we do something like (2bits + 6 bits) 1byte + (x + y - 6) bits pack, or do we use a bigger pack of lets 32 bits, but we can still fall into this issue ?

[hst-m]

Strings use Utf8 where each character is 1 byte/8 bits, so there is no benefit to store strings in packed bytes, you should just use the standard buf.write for strings and put it next to the packed bytes not inside. But for example if you map 'tatamotors' string to integer 0, and map 'tesla' string to integer 1, now your field only requires 1 bit to store all possible values, whereas 'tatamotors' string would have required 80 bits to store in Utf8, that is 80x size savings

By Double you mean the Double Floats which use 8 bytes/64 bits to store a floating point number, floating point numbers should also not be stored inside packed bytes, they need the full 64 bits. They can be stored next to packed bytes in the same Buffer. However your example number 35.67 instead of being stored as a floating point using 32 or 64 bits it could be stored as a 12 bit integer, pack it as integer 3567 and then divide by 100 after unpacking, this is why currency such as the Dollar $ is usually stored as Cents, so they are stored as integers vs floating point numbers which require much more space

bits is the minimum number of bits required to represent a value ?

Bits are the total number of bits needed to store all possible values for that field, as small as possible to hold all values. For example if a field could have an integer value in the range of 3010 to 3017, you might think to store the values 3010-3017 which would require 12 bits, but there are only 8 possible values which only require 3 bits, so instead store the values 0-7 and after unpacking map it back to original values

we can't pack this second data into 1 byte, so do we do something like (2bits + 6 bits) 1byte + (x + y - 6) bits pack, or do we use a bigger pack of lets 32 bits

You can do 8 bit pack with writeUInt8 or 16 bit pack with writeUInt16BE or 32 bit with writeUInt32BE

[hst-m]

Here is 32 bit example you can run in browser console:

const packBytes=(...bitsVals)=>bitsVals.reduce((byte,val,i)=>i%2?byte+val:byte<<val,0)>>>0
const unpackBytes=(byte,...bits)=>{for(let i=bits.length-1;i>=0;i--){let val=byte%(1<<bits[i]);byte>>>=bits[i];bits[i]=val}return bits}
data={
   field1:true, // 1 bit
   field2:false, // 1 bit
   field3:3, // 2 bits
   field4:7, // 3 bits
   field5:15, // 4 bits
   field6:31, // 5 bits
   field7:63, // 6 bits
   field8:1023, // 10 bits
} // total 32 bits
packed=packBytes(
   1,data.field1,
   1,data.field2,
   2,data.field3,
   3,data.field4,
   4,data.field5,
   5,data.field6,
   6,data.field7,
   10,data.field8
) // packed == 3221225471; packed.toString(2) == '10111111111111111111111111111111'
// node.js: buf.writeUInt32BE(packed)
// browser: packed=dataView.getUint32()
values=unpackBytes(packed,1,1,2,3,4,5,6,10) // values == [1, 0, 3, 7, 15, 31, 63, 1023]

This example uses 2x less bytes than avsc, 17x less bytes than msgpack, and 26x less bytes than json

Encoding	time (ns)	bytes
packBytes	95	4
avsc	255	9
msgpack	1750	67
json	1035	102

[hst-m]

I have performance improvement for this, I was looking at JS bitwise operations and found trick to convert the 32 bit signed integer to unsigned integer which allows full 32 bit operations instead of stopping at 31 bits, so 32 bit packBytes is now 5 times faster and can eliminate the separate 32 bit function, I updated the functions and examples

[rpgehlot]

Hello, is it possible under any scenario for uws to drop any message for any client even if i have closeOnBackPressureLimit set to true ?

[hst-m]

closeOnBackpressureLimit=true should not skip any messages for send or publish, messages get dropped when connection drops

[arupadhy]

Putting a question here since this thread seems to be relevant. In my use case, i would like the backpressure logic to dicatate the speed of the source (lets say some large database query that drives data flow) that is feeding into the socket as well, can i use 'closeOnBackpressureLimit' in conjunction with 'drain' event. iow, i would like to use ws.getBufferedAmount() < $myConfiguredValue check to slow down my source (db stream) but i still want library to close the socket instead of dropping the packet in the worst case.

[hst-m]

can i use 'closeOnBackpressureLimit' in conjunction with 'drain' event

Yes

[arupadhy]

Thanks for quick reply. docs didn't specify one way or the other. So just to make sure i understand, if i use closeOnBackpressureLimit: true, 'drain' event won't be fired at all... Instead socket will be dropped when backpressure goes over the limit...is that correct?

[hst-m]

Drain event still fires on every drain