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

Bunch 'o experimental changes

Browse Source
This commit is contained in:
Alex Hultman
2023-05-07 01:28:07 +02:00
parent 6f7d8657a4
commit abb22793ff
8 changed files with 220 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
benchmarks/Makefile
+1
View File
@@ -1,4 +1,5 @@
default:
g++ -flto -march=native parser.cpp -O3 -I../uSockets/src -o parser
clang -flto -O3 -DLIBUS_USE_OPENSSL -I../uSockets/src ../uSockets/src/*.c ../uSockets/src/eventing/*.c ../uSockets/src/crypto/*.c broadcast_test.c load_test.c scale_test.c -c
clang++ -flto -O3 -DLIBUS_USE_OPENSSL -I../uSockets/src ../uSockets/src/crypto/*.cpp -c -std=c++17
clang++ -flto -O3 -DLIBUS_USE_OPENSSL `ls *.o | grep -Ev "load_test|scale_test"` -lssl -lcrypto -o broadcast_test
benchmarks/README.md
+1 -2
View File
@@ -1,5 +1,4 @@
# Benchmark-driven development
Making decisions based on scientific benchmarking **while** you develop can guide you to create very efficient solutions if you have the dicipline to follow through. µWebSockets performs with **98%** the theoretical maximum for any user space Linux process - if anything would ever be faster, it would only be so by less than 2%. We know of no such project.
Http | WebSockets
@@ -32,6 +31,6 @@ Contrary to popular belief, "hello world benchmarks" are the most accurate and r
* Chatting (memory overhead)
* Notifications (memory overhead)
Most business applications of the above mentioned categories are implemented without a central on-disk DB, blocking or severely limiting hot-path performance. As such, web IO becomes a significant part of overall bottleneck, if not the only bottleneck.
Most business applications of the above mentioned categories are implemented without a central on-disk DB, blocking or severely limiting hot-path performance. As such, web IO becomes a significant part of overall bottleneck, if not the only bottleneck. Message echoing of around 1-16 kB or even as small as 512 bytes is a good test of the overall server plumbing (receive -> timeout clear -> emit to app -> timeout set -> send) for these applications.
Of course, if you build an app that *absolutely must* have an on-disk SQL DB central to all hot-paths, then µWebSockets is not the right tool for your app. Keep in mind that, finding a case where µWebSockets makes no difference, does not mean µWebSockets never makes a difference.
benchmarks/load_test.c
+32 -6
View File
@@ -1,6 +1,6 @@
/* This is a simple yet efficient WebSocket server benchmark much like WRK */
#define _BSD_SOURCE
#define _DEFAULT_SOURCE
#ifdef __APPLE__
#include <libkern/OSByteOrder.h>
@@ -62,6 +62,7 @@ void init_big_message(unsigned int size) {
web_socket_request_text_size = size + 6 + 8;
web_socket_request_text = malloc(web_socket_request_text_size);
memset(web_socket_request_text, 'T', web_socket_request_text_size);
web_socket_request_text[0] = 130;
web_socket_request_text[1] = 255;
uint64_t msg_size = htobe64(size);
@@ -72,6 +73,26 @@ void init_big_message(unsigned int size) {
web_socket_request_text[10] = 4;
}
void init_medium_message(unsigned int size) {
if (size > 65536) {
printf("Error: message size must be smaller\n");
exit(0);
}
web_socket_request_text_size = size + 6 + 2; // 8 for big
web_socket_request_text = malloc(web_socket_request_text_size);
memset(web_socket_request_text, 'T', web_socket_request_text_size);
web_socket_request_text[0] = 130;
web_socket_request_text[1] = 254;
uint16_t msg_size = htobe16(size);
memcpy(&web_socket_request_text[2], &msg_size, 2);
web_socket_request_text[4] = 1;
web_socket_request_text[5] = 2;
web_socket_request_text[6] = 3;
web_socket_request_text[7] = 4;
}
char request_deflate[] = "GET / HTTP/1.1\r\n"
"Upgrade: websocket\r\n"
"Connection: Upgrade\r\n"
@@ -229,7 +250,7 @@ int main(int argc, char **argv) {
/* Parse host and port */
if (argc != 6 && argc != 7) {
printf("Usage: connections host port ssl deflate [size_mb]\n");
printf("Usage: connections host port ssl deflate [size_kb]\n");
return 0;
}
@@ -248,10 +269,15 @@ int main(int argc, char **argv) {
} else {
/* Only if we are NOT using defalte can we support testing with 100mb for now */
if (argc == 7) {
int size_mb = atoi(argv[6]);
printf("Using message size of %d MB\n", size_mb);
/* Size has to be in MB since the minimal size is 64kb */
init_big_message(size_mb * 1024 * 1024);
int size_kb = atoi(argv[6]);
printf("Using message size of %d kB\n", size_kb);
/* Size has to be in KB since the minimal size for medium is 1kb */
if (size_kb <= 64) {
init_medium_message(size_kb * 1024);
} else {
init_big_message(size_kb * 1024);
}
}
web_socket_request = web_socket_request_text;
benchmarks/parser.cpp
+136
View File
@@ -0,0 +1,136 @@
/* This is a fuzz test of the websocket parser */
#define WIN32_EXPORT
/* We test the websocket parser */
#include "../src/WebSocketProtocol.h"
unsigned int messages = 0;
struct Impl {
static bool refusePayloadLength(uint64_t length, uWS::WebSocketState<true> *wState, void *s) {
/* We need a limit */
if (length > 16000) {
return true;
}
/* Return ok */
return false;
}
static bool setCompressed(uWS::WebSocketState<true> *wState, void *s) {
/* We support it */
return true;
}
static void forceClose(uWS::WebSocketState<true> *wState, void *s, std::string_view reason = {}) {
}
static bool handleFragment(char *data, size_t length, unsigned int remainingBytes, int opCode, bool fin, uWS::WebSocketState<true> *webSocketState, void *s) {
if (opCode == uWS::TEXT) {
if (!uWS::protocol::isValidUtf8((unsigned char *)data, length)) {
/* Return break */
return true;
}
} else if (opCode == uWS::CLOSE) {
uWS::protocol::parseClosePayload((char *)data, length);
}
messages += 1;
/* Return ok */
return false;
}
};
#include <iostream>
int web_socket_request_text_size;
char *web_socket_request_text;
void init_medium_message(unsigned int size) {
if (size > 65536) {
printf("Error: message size must be smaller\n");
exit(0);
}
web_socket_request_text_size = size + 6 + 2; // 8 for big
web_socket_request_text = ((char *) malloc(32 + web_socket_request_text_size + 32)) + 32;
memset(web_socket_request_text, 'T', web_socket_request_text_size + 32);
web_socket_request_text[0] = 130;
web_socket_request_text[1] = 254;
uint16_t msg_size = htobe16(size);
memcpy(&web_socket_request_text[2], &msg_size, 2);
web_socket_request_text[4] = 1;
web_socket_request_text[5] = 2;
web_socket_request_text[6] = 3;
web_socket_request_text[7] = 4;
}
int main() {
init_medium_message(1024);
/* Create the parser state */
uWS::WebSocketState<true> state;
unsigned char pre[32];
unsigned char web_socket_request_text_small[26] = {130, 128 | 20, 1, 2, 3, 4};
unsigned char post[32];
uint16_t msg_size = htobe16(1024);
{
clock_t start = clock();
for (unsigned long long i = 0; i < 100000000; i++) {
web_socket_request_text[0] = 130;
web_socket_request_text[1] = 254;
memcpy(&web_socket_request_text[2], &msg_size, 2);
web_socket_request_text[4] = 1;
web_socket_request_text[5] = 2;
web_socket_request_text[6] = 3;
web_socket_request_text[7] = 4;
// here we can either consume the whole message or consume the whole message minus 1 byte, causing a different path to be taken
uWS::WebSocketProtocol<true, Impl>::consume((char *) web_socket_request_text, web_socket_request_text_size-1, &state, nullptr);
}
clock_t stop = clock();
float seconds = ((float)(stop-start)/CLOCKS_PER_SEC);
std::cout << std::fixed << "Parsed incomplete 1 kB messages per second: " << ((float)messages / seconds) << std::endl;
}
{
messages = 0;
clock_t start = clock();
for (unsigned long long i = 0; i < 100000000; i++) {
web_socket_request_text[0] = 130;
web_socket_request_text[1] = 254;
memcpy(&web_socket_request_text[2], &msg_size, 2);
web_socket_request_text[4] = 1;
web_socket_request_text[5] = 2;
web_socket_request_text[6] = 3;
web_socket_request_text[7] = 4;
// here we can either consume the whole message or consume the whole message minus 1 byte, causing a different path to be taken
uWS::WebSocketProtocol<true, Impl>::consume((char *) web_socket_request_text, web_socket_request_text_size, &state, nullptr);
}
clock_t stop = clock();
float seconds = ((float)(stop-start)/CLOCKS_PER_SEC);
std::cout << std::fixed << "Parsed complete 1 kB messages per second: " << ((float)messages / seconds) << std::endl;
}
return 0;
}
build.c
+1 -1
View File
@@ -12,7 +12,7 @@ int main(int argc, char **argv) {
char *EXAMPLE_FILES[] = {"Http3Server", "Broadcast", "HelloWorld", "Crc32", "ServerName",
"EchoServer", "BroadcastingEchoServer", "UpgradeSync", "UpgradeAsync"};
strcat(CXXFLAGS, " -O3 -Wpedantic -Wall -Wextra -Wsign-conversion -Wconversion -std=c++20 -Isrc -IuSockets/src");
strcat(CXXFLAGS, " -march=native -O3 -Wpedantic -Wall -Wextra -Wsign-conversion -Wconversion -std=c++20 -Isrc -IuSockets/src");
strcat(LDFLAGS, " uSockets/*.o");
// By default we use LTO, but Windows does not support it
src/HttpResponse.h
+3 -1
View File
@@ -55,7 +55,9 @@ private:
/* Write an unsigned 32-bit integer in hex */
void writeUnsignedHex(unsigned int value) {
char buf[10];
/* Buf really only needs to be 8 long but building with
* -mavx2, GCC still wants to overstep it so made it 16 */
char buf[16];
int length = utils::u32toaHex(value, buf);
/* For now we do this copy */
src/Utilities.h
+2 -2
View File
@@ -30,8 +30,8 @@ inline int u32toaHex(uint32_t value, char *dst) {
char temp[10];
char *p = temp;
do {
*p++ = palette[value % 16];
value /= 16;
*p++ = palette[value & 15];
value >>= 4;
} while (value > 0);
int ret = (int) (p - temp);
src/WebSocketProtocol.h
+44 -17
View File
@@ -281,18 +281,42 @@ protected:
data[N - 1] ^= mask[(N - 1) % 4];
}
static inline void unmaskImprecise(char *dst, char *src, char *mask, unsigned int length) {
for (unsigned int n = (length >> 2) + 1; n; n--) {
*(dst++) = *(src++) ^ mask[0];
*(dst++) = *(src++) ^ mask[1];
*(dst++) = *(src++) ^ mask[2];
*(dst++) = *(src++) ^ mask[3];
template <int DESTINATION>
static inline void unmaskImprecise8(char *src, uint64_t mask, unsigned int length) {
for (unsigned int n = (length >> 3) + 1; n; n--) {
uint64_t loaded;
memcpy(&loaded, src, 8);
loaded ^= mask;
memcpy(src - DESTINATION, &loaded, 8);
src += 8;
}
}
static inline void unmaskImpreciseCopyMask(char *dst, char *src, char *maskPtr, unsigned int length) {
char mask[4] = {maskPtr[0], maskPtr[1], maskPtr[2], maskPtr[3]};
unmaskImprecise(dst, src, mask, length);
/* DESTINATION = 6 makes this not SIMD, DESTINATION = 4 is with SIMD but we don't want that for short messages */
template <int DESTINATION>
static inline void unmaskImprecise4(char *src, uint32_t mask, unsigned int length) {
for (unsigned int n = (length >> 2) + 1; n; n--) {
uint32_t loaded;
memcpy(&loaded, src, 4);
loaded ^= mask;
memcpy(src - DESTINATION, &loaded, 4);
src += 4;
}
}
template <int HEADER_SIZE>
static inline void unmaskImpreciseCopyMask(char *src, unsigned int length) {
if constexpr (HEADER_SIZE != 6) {
char mask[8] = {src[-4], src[-3], src[-2], src[-1], src[-4], src[-3], src[-2], src[-1]};
uint64_t maskInt;
memcpy(&maskInt, mask, 8);
unmaskImprecise8<HEADER_SIZE>(src, maskInt, length);
} else {
char mask[4] = {src[-4], src[-3], src[-2], src[-1]};
uint32_t maskInt;
memcpy(&maskInt, mask, 4);
unmaskImprecise4<HEADER_SIZE>(src, maskInt, length);
}
}
static inline void rotateMask(unsigned int offset, char *mask) {
@@ -332,9 +356,11 @@ protected:
}
if (payLength + MESSAGE_HEADER <= length) {
bool fin = isFin(src);
if (isServer) {
unmaskImpreciseCopyMask(src + MESSAGE_HEADER - 4, src + MESSAGE_HEADER, src + MESSAGE_HEADER - 4, (unsigned int) payLength);
if (Impl::handleFragment(src + MESSAGE_HEADER - 4, payLength, 0, wState->state.opCode[wState->state.opStack], isFin(src), wState, user)) {
/* This guy can never be assumed to be perfectly aligned since we can get multiple messages in one read */
unmaskImpreciseCopyMask<MESSAGE_HEADER>(src + MESSAGE_HEADER, (unsigned int) payLength);
if (Impl::handleFragment(src, payLength, 0, wState->state.opCode[wState->state.opStack], fin, wState, user)) {
return true;
}
} else {
@@ -343,7 +369,7 @@ protected:
}
}
if (isFin(src)) {
if (fin) {
wState->state.opStack--;
}
@@ -356,14 +382,15 @@ protected:
wState->state.wantsHead = false;
wState->remainingBytes = (unsigned int) (payLength - length + MESSAGE_HEADER);
bool fin = isFin(src);
if (isServer) {
if constexpr (isServer) {
memcpy(wState->mask, src + MESSAGE_HEADER - 4, 4);
unmaskImprecise(src, src + MESSAGE_HEADER, wState->mask, length - MESSAGE_HEADER);
uint64_t mask;
memcpy(&mask, src + MESSAGE_HEADER - 4, 4);
memcpy(((char *)&mask) + 4, src + MESSAGE_HEADER - 4, 4);
unmaskImprecise8<0>(src + MESSAGE_HEADER, mask, length);
rotateMask(4 - (length - MESSAGE_HEADER) % 4, wState->mask);
} else {
src += MESSAGE_HEADER;
}
Impl::handleFragment(src, length - MESSAGE_HEADER, wState->remainingBytes, wState->state.opCode[wState->state.opStack], fin, wState, user);
Impl::handleFragment(src + MESSAGE_HEADER, length - MESSAGE_HEADER, wState->remainingBytes, wState->state.opCode[wState->state.opStack], fin, wState, user);
return true;
}
}