Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 4 additions & 4 deletions packages/bun-uws/src/AsyncSocket.h
Original file line number Diff line number Diff line change
Expand Up @@ -183,8 +183,8 @@ struct AsyncSocket {

/* Returns the user space backpressure. */
size_t getBufferedAmount() {
/* We return the actual amount of bytes in backbuffer, including pendingRemoval */
return getAsyncSocketData()->buffer.totalLength();
/* Unsent bytes only; already-written bytes waiting for compaction are not backpressure. */
return getAsyncSocketData()->buffer.length();
}
Comment thread
robobun marked this conversation as resolved.

/* Returns the text representation of an IPv4 or IPv6 address */
Expand Down Expand Up @@ -253,7 +253,7 @@ struct AsyncSocket {
/* Check if we couldn't write the entire buffer */
if ((unsigned int) written < buffer_len) {
/* Remove the successfully written data from the buffer */
asyncSocketData->buffer.erase((unsigned int) written);
asyncSocketData->buffer.erase((size_t) written);

/* If we wrote less than we attempted, the socket buffer is likely full
* likely is used as an optimization hint to the compiler
Expand Down Expand Up @@ -301,7 +301,7 @@ struct AsyncSocket {
/* On failure return, otherwise continue down the function */
if ((unsigned int) written < buffer_len) {
/* Update buffering (todo: we can do better here if we keep track of what happens to this guy later on) */
asyncSocketData->buffer.erase((unsigned int) written);
asyncSocketData->buffer.erase((size_t) written);

if (optionally) {
/* Thankfully we can exit early here */
Expand Down
10 changes: 5 additions & 5 deletions packages/bun-uws/src/AsyncSocketData.h
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,7 @@ namespace uWS {

struct BackPressure {
std::string buffer;
unsigned int pendingRemoval = 0;
size_t pendingRemoval = 0;
BackPressure(BackPressure &&other) {
buffer = std::move(other.buffer);
pendingRemoval = other.pendingRemoval;
Expand All @@ -34,12 +34,12 @@ struct BackPressure {
void append(const char *data, size_t length) {
buffer.append(data, length);
}
void erase(unsigned int length) {
void erase(size_t length) {
pendingRemoval += length;
/* Always erase a minimum of 1/32th the current backpressure */
if (pendingRemoval > (buffer.length() >> 5)) {
/* Compact once half the buffer is dead space so draining n bytes moves
* O(n) total (geometric series). clear() releases capacity on full drain. */
if (pendingRemoval > (buffer.length() >> 1)) {
buffer.erase(0, pendingRemoval);
buffer.shrink_to_fit();
pendingRemoval = 0;
}
}
Expand Down
12 changes: 12 additions & 0 deletions packages/bun-uws/src/HttpResponse.h
Original file line number Diff line number Diff line change
Expand Up @@ -596,6 +596,18 @@ struct HttpResponse : public AsyncSocket<SSL> {

size_t length = data.length();

/* A write this large is split into several AsyncSocket::write calls below
* (chunk framing, INT_MAX slicing). Reserve once so those appends never
* realloc + copy a multi-GB backpressure buffer. */
if (length > 1024 * 1024) {
auto &bp = Super::getAsyncSocketData()->buffer;
size_t need = bp.totalLength() + length + 32;
if (need > bp.buffer.capacity()) {
/* libc++ reserve() is exact-fit; keep geometric growth explicit. */
bp.buffer.reserve(std::max(bp.buffer.capacity() * 2, need));
}
}
Comment thread
robobun marked this conversation as resolved.
Comment thread
robobun marked this conversation as resolved.

// Special handling for extremely large data (greater than UINT_MAX bytes)
// most clients expect a max of UINT_MAX, so we need to split the write into multiple writes
if (length > UINT_MAX) {
Expand Down
3 changes: 2 additions & 1 deletion packages/bun-uws/src/WebSocket.h
Original file line number Diff line number Diff line change
Expand Up @@ -106,7 +106,8 @@ struct WebSocket : AsyncSocket<SSL> {
}

size_t memoryCost() {
return getBufferedAmount() + sizeof(WebSocket);
/* Allocation footprint for reportExtraMemoryAllocated, not unsent bytes. */
return Super::getAsyncSocketData()->buffer.totalLength() + sizeof(WebSocket);
}

/* Sending fragmented messages puts a bit of effort on the user; you must not interleave regular sends
Expand Down
7 changes: 2 additions & 5 deletions test/js/node/http/node-http-backpressure-max.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@ import http from "node:http";
import type { AddressInfo } from "node:net";

describe("backpressure", () => {
// Linux CI only have 8GB with is not enought because we will clone all or most of this 4GB into memory
// Linux CI only has 8GB which is not enough because we clone all or most of this 4GB into memory.
it.skipIf(isCI && isLinux)(
Comment thread
coderabbitai[bot] marked this conversation as resolved.
"should handle backpressure with the maximum allowed bytes",
async () => {
Expand Down Expand Up @@ -45,9 +45,6 @@ describe("backpressure", () => {

expect(totalBytes).toBe(payloadSize);
},
// Moving 4 GiB through the server and the fetch reader takes ~60s on the
// slowest CI runners (darwin x64), which sat exactly at the old 60s
// limit and made the test flaky there.
120_000,
30_000,
);
});
Loading