talksik/uWebSockets
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add large non-ssl, non-compressed WebSocket message sending optimization

Browse Source
This commit is contained in:
Alex Hultman
2023-05-11 11:18:58 +02:00
parent c9724db3cf
commit f6f5a3d6e8
2 changed files with 54 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/WebSocket.h
+53 -34
View File
@@ -105,48 +105,67 @@ public:
/* If we are subscribers and have messages to drain we need to drain them here to stay synced */ /* If we are subscribers and have messages to drain we need to drain them here to stay synced */
WebSocketData *webSocketData = (WebSocketData *) Super::getAsyncSocketData(); WebSocketData *webSocketData = (WebSocketData *) Super::getAsyncSocketData();
if (webSocketData->subscriber) {
/* This will call back into us, send. */
webSocketContextData->topicTree->drain(webSocketData->subscriber);
}
/* Transform the message to compressed domain if requested */ /* Special path for long sends of non-compressed, non-SSL messages */
if (compress) { if (message.length() >= 16 * 1024 && !compress && !SSL && !webSocketData->subscriber && getBufferedAmount() == 0 && Super::getLoopData()->corkOffset == 0) {
WebSocketData *webSocketData = (WebSocketData *) Super::getAsyncSocketData(); char header[14];
int header_length = (int) protocol::formatMessage<isServer>(header, nullptr, 0, opCode, message.length(), compress, fin);
/* Check and correct the compress hint. It is never valid to compress 0 bytes */ int written = us_socket_write2(0, (struct us_socket_t *)this, header, header_length, message.data(), (int) message.length());
if (message.length() && opCode < 3 && webSocketData->compressionStatus == WebSocketData::ENABLED) {
LoopData *loopData = Super::getLoopData(); if (written != header_length + (int) message.length()) {
/* Compress using either shared or dedicated deflationStream */ /* Buffer up backpressure */
if (webSocketData->deflationStream) { if (written > header_length) {
message = webSocketData->deflationStream->deflate(loopData->zlibContext, message, false); webSocketData->buffer.append(message.data() + written, message.length() - (size_t) written);
} else { } else {
message = loopData->deflationStream->deflate(loopData->zlibContext, message, true); webSocketData->buffer.append(header + written, (size_t) header_length - (size_t) written);
} }
} else {
compress = false;
} }
} } else {
/* Get size, allocate size, write if needed */ if (webSocketData->subscriber) {
size_t messageFrameSize = protocol::messageFrameSize(message.length()); /* This will call back into us, send. */
auto [sendBuffer, sendBufferAttribute] = Super::getSendBuffer(messageFrameSize); webSocketContextData->topicTree->drain(webSocketData->subscriber);
protocol::formatMessage<isServer>(sendBuffer, message.data(), message.length(), opCode, message.length(), compress, fin); }
/* Depending on size of message we have different paths */ /* Transform the message to compressed domain if requested */
if (sendBufferAttribute == SendBufferAttribute::NEEDS_DRAIN) { if (compress) {
/* This is a drain */ WebSocketData *webSocketData = (WebSocketData *) Super::getAsyncSocketData();
auto[written, failed] = Super::write(nullptr, 0);
if (failed) { /* Check and correct the compress hint. It is never valid to compress 0 bytes */
/* Return false for failure, skipping to reset the timeout below */ if (message.length() && opCode < 3 && webSocketData->compressionStatus == WebSocketData::ENABLED) {
return BACKPRESSURE; LoopData *loopData = Super::getLoopData();
/* Compress using either shared or dedicated deflationStream */
if (webSocketData->deflationStream) {
message = webSocketData->deflationStream->deflate(loopData->zlibContext, message, false);
} else {
message = loopData->deflationStream->deflate(loopData->zlibContext, message, true);
}
} else {
compress = false;
}
} }
} else if (sendBufferAttribute == SendBufferAttribute::NEEDS_UNCORK) {
/* Uncork if we came here uncorked */ /* Get size, allocate size, write if needed */
auto [written, failed] = Super::uncork(); size_t messageFrameSize = protocol::messageFrameSize(message.length());
if (failed) { auto [sendBuffer, sendBufferAttribute] = Super::getSendBuffer(messageFrameSize);
return BACKPRESSURE; protocol::formatMessage<isServer>(sendBuffer, message.data(), message.length(), opCode, message.length(), compress, fin);
/* Depending on size of message we have different paths */
if (sendBufferAttribute == SendBufferAttribute::NEEDS_DRAIN) {
/* This is a drain */
auto[written, failed] = Super::write(nullptr, 0);
if (failed) {
/* Return false for failure, skipping to reset the timeout below */
return BACKPRESSURE;
}
} else if (sendBufferAttribute == SendBufferAttribute::NEEDS_UNCORK) {
/* Uncork if we came here uncorked */
auto [written, failed] = Super::uncork();
if (failed) {
return BACKPRESSURE;
}
} }
} }
/* Every successful send resets the timeout */ /* Every successful send resets the timeout */
uSockets
+1 -1
Submodule uSockets updated: b950efd6b1...0866daff30