Improve large backpressure drain performance
This commit is contained in:
+1
-1
@@ -176,7 +176,7 @@ protected:
|
|||||||
if ((unsigned int) written < asyncSocketData->buffer.length()) {
|
if ((unsigned int) written < asyncSocketData->buffer.length()) {
|
||||||
|
|
||||||
/* Update buffering (todo: we can do better here if we keep track of what happens to this guy later on) */
|
/* Update buffering (todo: we can do better here if we keep track of what happens to this guy later on) */
|
||||||
asyncSocketData->buffer = asyncSocketData->buffer.substr((size_t) written);
|
asyncSocketData->buffer.erase((unsigned int) written);
|
||||||
|
|
||||||
if (optionally) {
|
if (optionally) {
|
||||||
/* Thankfully we can exit early here */
|
/* Thankfully we can exit early here */
|
||||||
|
|||||||
+39
-3
@@ -1,5 +1,5 @@
|
|||||||
/*
|
/*
|
||||||
* Authored by Alex Hultman, 2018-2019.
|
* Authored by Alex Hultman, 2018-2021.
|
||||||
* Intellectual property of third-party.
|
* Intellectual property of third-party.
|
||||||
|
|
||||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||||
@@ -22,15 +22,51 @@
|
|||||||
|
|
||||||
namespace uWS {
|
namespace uWS {
|
||||||
|
|
||||||
|
struct BackPressure {
|
||||||
|
std::string buffer;
|
||||||
|
unsigned int pendingRemoval = 0;
|
||||||
|
BackPressure(BackPressure &&other) {
|
||||||
|
buffer = std::move(other.buffer);
|
||||||
|
pendingRemoval = other.pendingRemoval;
|
||||||
|
}
|
||||||
|
BackPressure() = default;
|
||||||
|
void append(const char *data, size_t length) {
|
||||||
|
buffer.append(data, length);
|
||||||
|
}
|
||||||
|
void erase(unsigned int length) {
|
||||||
|
pendingRemoval += length;
|
||||||
|
if (pendingRemoval > 1024 * 1024 || !(buffer.length() - pendingRemoval)) {
|
||||||
|
buffer.erase(0, pendingRemoval);
|
||||||
|
pendingRemoval = 0;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
size_t length() {
|
||||||
|
return buffer.length() - pendingRemoval;
|
||||||
|
}
|
||||||
|
void clear() {
|
||||||
|
pendingRemoval = 0;
|
||||||
|
buffer.clear();
|
||||||
|
}
|
||||||
|
void reserve(size_t length) {
|
||||||
|
buffer.reserve(length + pendingRemoval);
|
||||||
|
}
|
||||||
|
const char *data() {
|
||||||
|
return buffer.data() + pendingRemoval;
|
||||||
|
}
|
||||||
|
size_t size() {
|
||||||
|
return length();
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
/* Depending on how we want AsyncSocket to function, this will need to change */
|
/* Depending on how we want AsyncSocket to function, this will need to change */
|
||||||
|
|
||||||
template <bool SSL>
|
template <bool SSL>
|
||||||
struct AsyncSocketData {
|
struct AsyncSocketData {
|
||||||
/* This will do for now */
|
/* This will do for now */
|
||||||
std::string buffer;
|
BackPressure buffer;
|
||||||
|
|
||||||
/* Allow move constructing us */
|
/* Allow move constructing us */
|
||||||
AsyncSocketData(std::string &&backpressure) : buffer(std::move(backpressure)) {
|
AsyncSocketData(BackPressure &&backpressure) : buffer(std::move(backpressure)) {
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
+1
-1
@@ -273,7 +273,7 @@ public:
|
|||||||
HttpContext<SSL> *httpContext = (HttpContext<SSL> *) us_socket_context(SSL, (struct us_socket_t *) this);
|
HttpContext<SSL> *httpContext = (HttpContext<SSL> *) us_socket_context(SSL, (struct us_socket_t *) this);
|
||||||
|
|
||||||
/* Move any backpressure out of HttpResponse */
|
/* Move any backpressure out of HttpResponse */
|
||||||
std::string backpressure(std::move(((AsyncSocketData<SSL> *) getHttpResponseData())->buffer));
|
BackPressure backpressure(std::move(((AsyncSocketData<SSL> *) getHttpResponseData())->buffer));
|
||||||
|
|
||||||
/* Destroy HttpResponseData */
|
/* Destroy HttpResponseData */
|
||||||
getHttpResponseData()->~HttpResponseData();
|
getHttpResponseData()->~HttpResponseData();
|
||||||
|
|||||||
+1
-1
@@ -34,7 +34,7 @@ struct WebSocket : AsyncSocket<SSL> {
|
|||||||
private:
|
private:
|
||||||
typedef AsyncSocket<SSL> Super;
|
typedef AsyncSocket<SSL> Super;
|
||||||
|
|
||||||
void *init(bool perMessageDeflate, CompressOptions compressOptions, std::string &&backpressure) {
|
void *init(bool perMessageDeflate, CompressOptions compressOptions, BackPressure &&backpressure) {
|
||||||
new (us_socket_ext(SSL, (us_socket_t *) this)) WebSocketData(perMessageDeflate, compressOptions, std::move(backpressure));
|
new (us_socket_ext(SSL, (us_socket_t *) this)) WebSocketData(perMessageDeflate, compressOptions, std::move(backpressure));
|
||||||
return this;
|
return this;
|
||||||
}
|
}
|
||||||
|
|||||||
+1
-1
@@ -50,7 +50,7 @@ private:
|
|||||||
/* We could be a subscriber */
|
/* We could be a subscriber */
|
||||||
Subscriber *subscriber = nullptr;
|
Subscriber *subscriber = nullptr;
|
||||||
public:
|
public:
|
||||||
WebSocketData(bool perMessageDeflate, CompressOptions compressOptions, std::string &&backpressure) : AsyncSocketData<false>(std::move(backpressure)), WebSocketState<true>() {
|
WebSocketData(bool perMessageDeflate, CompressOptions compressOptions, BackPressure &&backpressure) : AsyncSocketData<false>(std::move(backpressure)), WebSocketState<true>() {
|
||||||
compressionStatus = perMessageDeflate ? ENABLED : DISABLED;
|
compressionStatus = perMessageDeflate ? ENABLED : DISABLED;
|
||||||
|
|
||||||
/* Initialize the dedicated sliding window */
|
/* Initialize the dedicated sliding window */
|
||||||
|
|||||||
Reference in New Issue
Block a user