Pass Autobahn with restored WS perf.
This commit is contained in:
+21
-15
@@ -68,25 +68,31 @@ protected:
|
|||||||
|
|
||||||
/* Cork this socket. Only one socket may ever be corked per-loop at any given time */
|
/* Cork this socket. Only one socket may ever be corked per-loop at any given time */
|
||||||
void cork() {
|
void cork() {
|
||||||
getLoopData()->corked = true;
|
/* What if another socket is corked? */
|
||||||
|
getLoopData()->corkedSocket = this;
|
||||||
}
|
}
|
||||||
|
|
||||||
// this is highly broken right now, should properly make use of secondary buffer if needed
|
/* Returns wheter we are corked or not */
|
||||||
|
bool isCorked() {
|
||||||
|
return getLoopData()->corkedSocket == this;
|
||||||
|
}
|
||||||
|
|
||||||
|
/* Returns a suitable buffer for temporary assemblation of send data */
|
||||||
std::pair<char *, bool> getSendBuffer(size_t size) {
|
std::pair<char *, bool> getSendBuffer(size_t size) {
|
||||||
|
/* If we are corked and we have room, return the cork buffer itself */
|
||||||
// for now, just return this straight up
|
|
||||||
|
|
||||||
LoopData *loopData = getLoopData();
|
LoopData *loopData = getLoopData();
|
||||||
|
if (loopData->corkedSocket == this && loopData->corkOffset + size < LoopData::CORK_BUFFER_SIZE) {
|
||||||
|
char *sendBuffer = loopData->corkBuffer + loopData->corkOffset;
|
||||||
|
loopData->corkOffset += size;
|
||||||
|
return {sendBuffer, false};
|
||||||
|
} else {
|
||||||
|
// slow path for now
|
||||||
|
|
||||||
char *sendBuffer = loopData->corkBuffer + loopData->corkOffset;
|
return {(char *) malloc(size), true};
|
||||||
|
|
||||||
|
// if we are out of buffer, fail this completely?
|
||||||
|
|
||||||
// very broken
|
}
|
||||||
loopData->corkOffset += size;
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
return {sendBuffer, false};
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/* Write in three levels of prioritization: cork-buffer, syscall, socket-buffer. Always drain if possible.
|
/* Write in three levels of prioritization: cork-buffer, syscall, socket-buffer. Always drain if possible.
|
||||||
@@ -126,7 +132,7 @@ protected:
|
|||||||
}
|
}
|
||||||
|
|
||||||
if (length) {
|
if (length) {
|
||||||
if (loopData->corked) {
|
if (loopData->corkedSocket == this) {
|
||||||
/* We are corked */
|
/* We are corked */
|
||||||
if (LoopData::CORK_BUFFER_SIZE - loopData->corkOffset >= length) {
|
if (LoopData::CORK_BUFFER_SIZE - loopData->corkOffset >= length) {
|
||||||
/* If the entire chunk fits in cork buffer */
|
/* If the entire chunk fits in cork buffer */
|
||||||
@@ -186,8 +192,8 @@ protected:
|
|||||||
std::pair<int, bool> uncork(const char *src = nullptr, int length = 0, bool optionally = false) {
|
std::pair<int, bool> uncork(const char *src = nullptr, int length = 0, bool optionally = false) {
|
||||||
LoopData *loopData = getLoopData();
|
LoopData *loopData = getLoopData();
|
||||||
|
|
||||||
if (loopData->corked) {
|
if (loopData->corkedSocket == this) {
|
||||||
loopData->corked = false;
|
loopData->corkedSocket = nullptr;
|
||||||
|
|
||||||
if (loopData->corkOffset) {
|
if (loopData->corkOffset) {
|
||||||
/* Corked data is already accounted for via its write call */
|
/* Corked data is already accounted for via its write call */
|
||||||
|
|||||||
+1
-1
@@ -44,7 +44,7 @@ public:
|
|||||||
/* Cork data */
|
/* Cork data */
|
||||||
char *corkBuffer = new char[CORK_BUFFER_SIZE];
|
char *corkBuffer = new char[CORK_BUFFER_SIZE];
|
||||||
int corkOffset = 0;
|
int corkOffset = 0;
|
||||||
bool corked = false;
|
void *corkedSocket = nullptr;
|
||||||
|
|
||||||
/* Compression data */
|
/* Compression data */
|
||||||
InflationStream *inflationStream = nullptr;
|
InflationStream *inflationStream = nullptr;
|
||||||
|
|||||||
+5
-1
@@ -39,7 +39,7 @@ private:
|
|||||||
}
|
}
|
||||||
public:
|
public:
|
||||||
|
|
||||||
// this function need clean-ups and perf. fixes
|
/* Send or buffer a WebSocket frame, compressed or not. Returns false on increased user space backpressure. */
|
||||||
bool send(std::string_view message, uWS::OpCode opCode = uWS::OpCode::BINARY, bool compress = false) {
|
bool send(std::string_view message, uWS::OpCode opCode = uWS::OpCode::BINARY, bool compress = false) {
|
||||||
/* Transform the message to compressed domain if requested */
|
/* Transform the message to compressed domain if requested */
|
||||||
if (compress) {
|
if (compress) {
|
||||||
@@ -53,6 +53,10 @@ public:
|
|||||||
WebSocketProtocol<isServer, WebSocket<SSL, isServer>>::formatMessage(sendBuffer, message.data(), message.length(), opCode, message.length(), false);
|
WebSocketProtocol<isServer, WebSocket<SSL, isServer>>::formatMessage(sendBuffer, message.data(), message.length(), opCode, message.length(), false);
|
||||||
if (requiresWrite) {
|
if (requiresWrite) {
|
||||||
auto[written, failed] = Super::write(sendBuffer, messageFrameSize);
|
auto[written, failed] = Super::write(sendBuffer, messageFrameSize);
|
||||||
|
|
||||||
|
/* For now, we are slow here (fix!) */
|
||||||
|
free(sendBuffer);
|
||||||
|
|
||||||
/* Return true for success */
|
/* Return true for success */
|
||||||
return !failed;
|
return !failed;
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -270,7 +270,7 @@ private:
|
|||||||
/* Handle WebSocket data streams */
|
/* Handle WebSocket data streams */
|
||||||
static_dispatch(us_ssl_socket_context_on_data, us_socket_context_on_data)(getSocketContext(), [](auto *s, char *data, int length) {
|
static_dispatch(us_ssl_socket_context_on_data, us_socket_context_on_data)(getSocketContext(), [](auto *s, char *data, int length) {
|
||||||
|
|
||||||
std::cout << "websocket data" << std::endl;
|
//std::cout << "websocket data" << std::endl;
|
||||||
|
|
||||||
/* We always cork on data */
|
/* We always cork on data */
|
||||||
AsyncSocket<SSL> *webSocket = (AsyncSocket<SSL> *) s;
|
AsyncSocket<SSL> *webSocket = (AsyncSocket<SSL> *) s;
|
||||||
|
|||||||
Reference in New Issue
Block a user