usockets: give the receive buffers' unused pages back to the system on Linux

packages/bun-usockets/src/crypto/openssl.c

@@ -749,6 +749,7 @@ void us_internal_init_loop_ssl_data(struct us_loop_t *loop) {
     if (!loop_ssl_data) Bun__outOfMemory();
     loop_ssl_data->ssl_read_output = ssl_alloc_read_output();
     if (!loop_ssl_data->ssl_read_output) Bun__outOfMemory();
+    us_internal_schedule_recv_buffer_release(loop);
 
     OPENSSL_init_ssl(0, NULL);
 
@@ -785,6 +786,11 @@ void us_internal_free_loop_ssl_data(struct us_loop_t *loop) {
   }
 }
 
+char *us_internal_loop_ssl_read_output(struct us_loop_t *loop) {
+  struct loop_ssl_data *loop_ssl_data = (struct loop_ssl_data *)loop->data.ssl_data;
+  return loop_ssl_data ? loop_ssl_data->ssl_read_output : NULL;
+}
+
 /* ── SSL_CTX construction ────────────────────────────────────────────────── */
 
 static int us_ssl_ctx_use_privatekey_content(SSL_CTX *ctx, const char *content, int type) {
@@ -2854,6 +2860,10 @@ struct us_socket_t *us_internal_ssl_on_data(struct us_socket_t *socket, char *da
       if (!s) return NULL;
     }
 
+    /* Not the ciphertext's length: a record that the previous read began adds to this one's plaintext. */
+    if (read <= LIBUS_RECV_BUFFER_RESIDENT_LENGTH && read + just_read > LIBUS_RECV_BUFFER_RESIDENT_LENGTH) {
+      us_internal_schedule_recv_buffer_release(s->group->loop);
+    }
     read += just_read;
 
     if (read == LIBUS_RECV_BUFFER_LENGTH) {

packages/bun-usockets/src/eventing/epoll_kqueue.c

@@ -428,20 +428,20 @@ static void us_internal_drain_ready_polls(struct us_loop_t *loop) {
     }
 }
 
-/* Bound `timeout` by the socket-timeout sweep deadline (NULL == forever). */
-static const struct timespec *us_internal_clamp_to_sweep(struct us_loop_t *loop, const struct timespec *timeout, struct timespec *storage) {
-    long long ns = us_internal_sweep_timeout_ns(loop);
+/* Bound `timeout` by the loop's own deadlines (NULL == forever). */
+static const struct timespec *us_internal_clamp_to_loop_timeout(struct us_loop_t *loop, const struct timespec *timeout, struct timespec *storage) {
+    long long ns = us_internal_loop_timeout_ns(loop);
     if (ns < 0) {
         return timeout;
     }
-    long long sweep_sec = ns / 1000000000LL;
-    long long sweep_nsec = ns % 1000000000LL;
-    if (timeout && (timeout->tv_sec < sweep_sec ||
-                    (timeout->tv_sec == sweep_sec && timeout->tv_nsec <= sweep_nsec))) {
+    long long sec = ns / 1000000000LL;
+    long long nsec = ns % 1000000000LL;
+    if (timeout && (timeout->tv_sec < sec ||
+                    (timeout->tv_sec == sec && timeout->tv_nsec <= nsec))) {
         return timeout;
     }
-    storage->tv_sec = (time_t) sweep_sec;
-    storage->tv_nsec = (long) sweep_nsec;
+    storage->tv_sec = (time_t) sec;
+    storage->tv_nsec = (long) nsec;
     return storage;
 }
 
@@ -452,8 +452,8 @@ void us_loop_run(struct us_loop_t *loop) {
         /* Emit pre callback */
         us_internal_loop_pre(loop);
 
-        struct timespec sweep_ts;
-        const struct timespec *timeout = us_internal_clamp_to_sweep(loop, NULL, &sweep_ts);
+        struct timespec loop_ts;
+        const struct timespec *timeout = us_internal_clamp_to_loop_timeout(loop, NULL, &loop_ts);
 
         /* Fetch ready polls */
 #ifdef LIBUS_USE_EPOLL
@@ -465,6 +465,7 @@ void us_loop_run(struct us_loop_t *loop) {
         us_internal_dispatch_ready_polls(loop);
         us_internal_drain_ready_polls(loop);
         us_internal_sweep_if_due(loop);
+        us_internal_release_recv_buffers_if_due(loop);
 
         /* Emit post callback */
         us_internal_loop_post(loop);
@@ -499,8 +500,8 @@ void us_loop_run_bun_tick(struct us_loop_t *loop, const struct timespec* timeout
         }
     }
 
-    struct timespec sweep_ts;
-    timeout = us_internal_clamp_to_sweep(loop, timeout, &sweep_ts);
+    struct timespec loop_ts;
+    timeout = us_internal_clamp_to_loop_timeout(loop, timeout, &loop_ts);
 
     const unsigned int had_wakeups = __atomic_exchange_n(&loop->pending_wakeups, 0, __ATOMIC_ACQUIRE);
     const int will_idle_inside_event_loop = had_wakeups == 0 && (!timeout || (timeout->tv_nsec != 0 || timeout->tv_sec != 0));
@@ -558,6 +559,7 @@ void us_loop_run_bun_tick(struct us_loop_t *loop, const struct timespec* timeout
     us_internal_dispatch_ready_polls(loop);
     us_internal_drain_ready_polls(loop);
     us_internal_sweep_if_due(loop);
+    us_internal_release_recv_buffers_if_due(loop);
 
     /* Emit post callback */
     us_internal_loop_post(loop);

packages/bun-usockets/src/internal/internal.h

@@ -209,9 +209,20 @@ void us_internal_disable_sweep_timer(struct us_loop_t *loop);
 /* CLOCK_MONOTONIC in ns. The clock every deadline on the loop is measured
  * against, so anything comparing against one must read it and not another. */
 uint64_t us_internal_monotonic_ns(void);
-long long us_internal_sweep_timeout_ns(struct us_loop_t *loop);
+/* ns until the loop's next deadline of its own (sweep, receive buffer release), or -1. */
+long long us_internal_loop_timeout_ns(struct us_loop_t *loop);
 void us_internal_sweep_if_due(struct us_loop_t *loop);
 #endif
+#ifdef __linux__
+/* Call after writing past LIBUS_RECV_BUFFER_RESIDENT_LENGTH in either receive buffer.
+ * Linux only: its MADV_DONTNEED frees anonymous pages at once and leaves nothing to undo before
+ * the next write. No other platform's madvise promises both. */
+void us_internal_schedule_recv_buffer_release(struct us_loop_t *loop);
+void us_internal_release_recv_buffers_if_due(struct us_loop_t *loop);
+#else
+#define us_internal_schedule_recv_buffer_release(loop) ((void) 0)
+#define us_internal_release_recv_buffers_if_due(loop) ((void) 0)
+#endif
 void us_internal_free_closed_sockets(us_loop_r loop);
 /* Defers closing `s` with `code` to us_internal_loop_post. */
 void us_internal_loop_close_unresumable_socket(struct us_loop_t *loop, struct us_socket_t *s, int code);
@@ -263,6 +274,8 @@ void us_internal_poll_close_fd(struct us_poll_t *p);
 /* SSL loop data */
 void us_internal_init_loop_ssl_data(us_loop_r loop);
 void us_internal_free_loop_ssl_data(us_loop_r loop);
+/* The loop's TLS plaintext buffer, sized like recv_buf. NULL before the loop's first TLS socket. */
+char *us_internal_loop_ssl_read_output(us_loop_r loop);
 
 /* Socket context related */
 void us_internal_socket_group_link_socket(us_socket_group_r group, us_socket_r s);

packages/bun-usockets/src/internal/loop_data.h

@@ -104,6 +104,10 @@ struct us_internal_loop_data_t {
     /* Filled by us_socket_resume(), closed by us_internal_loop_post(). */
     int num_unresumable_sockets;
     struct us_internal_unresumable_socket_t *unresumable_sockets;
+    /* Absolute monotonic ns at which the receive buffers give the pages past
+     * LIBUS_RECV_BUFFER_RESIDENT_LENGTH back to the system, or 0. Folded into
+     * the poll timeout. Only Linux sets it. */
+    long long recv_buf_release_at_ns;
 };
 
 #endif // LOOP_DATA_H

packages/bun-usockets/src/libusockets.h

@@ -93,6 +93,8 @@
 
 /* 512kb shared receive buffer */
 #define LIBUS_RECV_BUFFER_LENGTH 524288
+/* What a receive buffer keeps resident once its loop stops getting larger reads (Linux) */
+#define LIBUS_RECV_BUFFER_RESIDENT_LENGTH 65536
 
 /* Small 16KB shared send buffer for UDP packet metadata */
 #define LIBUS_SEND_BUFFER_LENGTH (1 << 14)
@@ -358,6 +360,12 @@ void us_socket_group_close_all_ex(us_socket_group_r group, int also_listeners) n
  * connections, 0 if the loop was already idle. Listen sockets are NOT closed
  * (see close_all_ex). */
 int us_loop_close_all_groups(us_loop_r loop) nonnull_fn_decl;
+#ifdef __linux__
+/* The whole pages past LIBUS_RECV_BUFFER_RESIDENT_LENGTH of the socket receive buffer and of the
+ * TLS plaintext buffer, which is empty before the loop's first TLS socket. */
+struct iovec;
+void us_loop_recv_buffer_tails(us_loop_r loop, struct iovec *tails /* [2] */) nonnull_fn_decl;
+#endif
 
 unsigned short us_socket_group_timestamp(us_socket_group_r group) nonnull_fn_decl;
 struct us_loop_t *us_socket_group_loop(us_socket_group_r group) nonnull_fn_decl __attribute((returns_nonnull));

packages/bun-usockets/src/loop.c

@@ -30,6 +30,8 @@
 #ifdef __linux__
 #include <netinet/in.h>
 #include <linux/errqueue.h>
+#include <sys/mman.h>
+#include <sys/uio.h>
 #endif
 
 #if __has_include("wtf/Platform.h")
@@ -95,13 +97,19 @@ void us_internal_disable_sweep_timer(struct us_loop_t *loop) {
     }
 }
 
-long long us_internal_sweep_timeout_ns(struct us_loop_t *loop) {
-    if (loop->data.sweep_next_tick_ns < 0) {
+long long us_internal_loop_timeout_ns(struct us_loop_t *loop) {
+    long long deadline = loop->data.sweep_next_tick_ns;
+    /* A nested tick does not release, so a deadline it cannot clear must not wake it. */
+    long long release_at = loop->data.tick_depth <= 1 ? loop->data.recv_buf_release_at_ns : 0;
+    if (release_at && (deadline < 0 || release_at < deadline)) {
+        deadline = release_at;
+    }
+    if (deadline < 0) {
         return -1;
     }
     /* Its own reading, deliberately: this bounds the poll so the sweep is not
      * starved, and a caller's older reading would round the deadline up. */
-    long long diff = loop->data.sweep_next_tick_ns - (long long) us_internal_monotonic_ns();
+    long long diff = deadline - (long long) us_internal_monotonic_ns();
     return diff > 0 ? diff : 0;
 }
 
@@ -120,6 +128,54 @@ void us_internal_sweep_if_due(struct us_loop_t *loop) {
 
 #endif
 
+#ifdef __linux__
+
+#define LIBUS_RECV_BUFFER_RELEASE_DELAY_NS 1000000000LL
+
+static struct iovec us_internal_recv_buffer_tail(char *buf) {
+    struct iovec tail = {0};
+    if (!buf) return tail;
+    uintptr_t page_size = (uintptr_t) sysconf(_SC_PAGESIZE);
+    uintptr_t begin = ((uintptr_t) buf + LIBUS_RECV_BUFFER_PADDING + LIBUS_RECV_BUFFER_RESIDENT_LENGTH + page_size - 1) & ~(page_size - 1);
+    uintptr_t end = ((uintptr_t) buf + LIBUS_RECV_BUFFER_LENGTH + LIBUS_RECV_BUFFER_PADDING * 2) & ~(page_size - 1);
+    if (end > begin) {
+        tail.iov_base = (void *) begin;
+        tail.iov_len = end - begin;
+    }
+    return tail;
+}
+
+void us_loop_recv_buffer_tails(struct us_loop_t *loop, struct iovec *tails) {
+    tails[0] = us_internal_recv_buffer_tail(loop->data.recv_buf);
+#ifndef LIBUS_NO_SSL
+    tails[1] = us_internal_recv_buffer_tail(us_internal_loop_ssl_read_output(loop));
+#else
+    tails[1] = us_internal_recv_buffer_tail(NULL);
+#endif
+}
+
+void us_internal_schedule_recv_buffer_release(struct us_loop_t *loop) {
+    loop->data.recv_buf_release_at_ns = (long long) us_internal_monotonic_ns() + LIBUS_RECV_BUFFER_RELEASE_DELAY_NS;
+}
+
+void us_internal_release_recv_buffers_if_due(struct us_loop_t *loop) {
+    /* The data callback that a nested tick runs under may still be reading its buffer. */
+    if (!loop->data.recv_buf_release_at_ns || loop->data.tick_depth > 1) {
+        return;
+    }
+    if ((long long) us_internal_monotonic_ns() < loop->data.recv_buf_release_at_ns) {
+        return;
+    }
+    loop->data.recv_buf_release_at_ns = 0;
+    struct iovec tails[2];
+    us_loop_recv_buffer_tails(loop, tails);
+    for (int i = 0; i < 2; i++) {
+        if (tails[i].iov_len) madvise(tails[i].iov_base, tails[i].iov_len, MADV_DONTNEED);
+    }
+}
+
+#endif
+
 
 /* -1 if the wakeup async cannot be created; nothing is left allocated in loop->data. */
 int us_internal_loop_data_init(struct us_loop_t *loop, void (*wakeup_cb)(struct us_loop_t *loop),
@@ -136,6 +192,8 @@ int us_internal_loop_data_init(struct us_loop_t *loop, void (*wakeup_cb)(struct
     /* Every read on this loop writes into recv_buf; a NULL here makes each one
      * fail with EFAULT for the life of the process. */
     if (!loop->data.recv_buf || !loop->data.send_buf) Bun__outOfMemory();
+    /* The allocator may have handed out pages that an earlier owner left resident. */
+    us_internal_schedule_recv_buffer_release(loop);
     loop->data.pre_cb = pre_cb;
     loop->data.post_cb = post_cb;
     loop->data.wakeup_async = us_internal_create_async(loop, 1, 0);
@@ -808,6 +866,7 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in
                     #endif
 
                     if (length > 0) {
+                        if (length > LIBUS_RECV_BUFFER_RESIDENT_LENGTH) us_internal_schedule_recv_buffer_release(loop);
                         s = s->ssl ? us_internal_ssl_on_data(s, loop->data.recv_buf + LIBUS_RECV_BUFFER_PADDING, length)
                                    : us_dispatch_data(s, loop->data.recv_buf + LIBUS_RECV_BUFFER_PADDING, length);
                         /* After socket adoption, track the new socket; the old one becomes invalid */
@@ -1099,6 +1158,8 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in
                     bsd_udp_setup_recvbuf(&recvbuf, u->loop->data.recv_buf, LIBUS_RECV_BUFFER_LENGTH);
                     int npackets = bsd_recvmmsg(us_poll_fd(p), &recvbuf, MSG_DONTWAIT, u->shared_fd ? 1 : LIBUS_UDP_RECV_COUNT);
                     if (npackets > 0) {
+                        /* Each packet has a LIBUS_UDP_MAX_SIZE slot of its own. */
+                        if (npackets > 1) us_internal_schedule_recv_buffer_release(u->loop);
                         u->on_data(u, &recvbuf, npackets);
                     } else {
                         if (npackets == LIBUS_SOCKET_ERROR) {

src/http/HTTPThread.rs

@@ -1408,6 +1408,24 @@ pub fn is_initialized() -> bool {
     crate::HTTP_THREAD_INIT.load(Ordering::Acquire)
 }
 
+/// For `bun:internal-for-testing`, on any thread. Null until the HTTP thread runs.
+#[unsafe(no_mangle)]
+pub extern "C" fn Bun__HTTPThread__uwsLoopForTesting() -> *mut uws::Loop {
+    if !is_initialized() {
+        return core::ptr::null_mut();
+    }
+    // SAFETY: same shape as `shutdown_for_exit()` below.
+    let thread = unsafe {
+        bun_ptr::ParentRef::<HttpThread>::from_raw(
+            (*crate::HTTP_THREAD.get_unchecked()).as_mut_ptr(),
+        )
+    };
+    if !thread.has_awoken.load(Ordering::Acquire) {
+        return core::ptr::null_mut();
+    }
+    thread.uws_loop
+}
+
 /// Called from `bun_jsc::VirtualMachine::global_exit()` on the JS thread,
 /// before `~VM`. Asks the HTTP daemon thread to reclaim every in-flight
 /// `ThreadlocalAsyncHTTP` box and waits (with a short timeout) for it to ack.

src/js/internal-for-testing.ts

@@ -942,6 +942,23 @@ export const isMemoryPressureWatcherInstalled: () => boolean = $newCppFunction(
   0,
 );
 
+// Linux: the part of the socket receive buffer and of the TLS plaintext buffer that an event loop gives back to
+// the system. This thread's loop, or the HTTP client thread's once it has served a request.
+export const recvBufferTails: (
+  ofHTTPThread?: boolean,
+) => Record<"recv" | "tls", { address: number; length: number }> | undefined = $newCppFunction(
+  "InternalForTesting.cpp",
+  "jsFunction_recvBufferTails",
+  1,
+);
+
+// Linux: mincore() over a page-aligned range of this process, any thread's. Negative errno on failure.
+export const residentBytes: (address: number, length: number) => number | undefined = $newCppFunction(
+  "InternalForTesting.cpp",
+  "jsFunction_residentBytes",
+  2,
+);
+
 // `parallelism` threads each spawn `iterations` no-op threads through bun's own pthread_create,
 // writing every failure to `fd`. Returns the first failing errno, or with `detach` 0 as soon as
 // every loop runs; a detached loop also writes to `fd` when it runs out of iterations.

src/jsc/bindings/InternalForTesting.cpp

@@ -23,6 +23,12 @@
 #include <sched.h>
 #include <unistd.h>
 #endif
+#if OS(LINUX)
+#include "libusockets.h"
+#include "_libusockets.h"
+#include <sys/mman.h>
+#include <sys/uio.h>
+#endif
 #include <JavaScriptCore/HeapProfiler.h>
 #include <JavaScriptCore/ObjectConstructor.h>
 #if ENABLE(AOT)
@@ -338,6 +344,58 @@ JSC_DEFINE_HOST_FUNCTION(jsFunction_isMemoryPressureWatcherInstalled, (JSC::JSGl
     return JSValue::encode(jsBoolean(Bun__MemoryPressure__isInstalled(defaultGlobalObject(globalObject))));
 }
 
+extern "C" struct us_loop_t* Bun__HTTPThread__uwsLoopForTesting();
+
+JSC_DEFINE_HOST_FUNCTION(jsFunction_recvBufferTails, (JSC::JSGlobalObject * globalObject, JSC::CallFrame* callFrame))
+{
+#if OS(LINUX)
+    struct us_loop_t* loop = callFrame->argument(0).toBoolean(globalObject) ? Bun__HTTPThread__uwsLoopForTesting() : uws_get_loop();
+    if (!loop)
+        return encodedJSUndefined();
+    struct iovec tails[2];
+    us_loop_recv_buffer_tails(loop, tails);
+    auto& vm = JSC::getVM(globalObject);
+    auto toJS = [&](const struct iovec& tail) {
+        JSObject* object = constructEmptyObject(globalObject);
+        object->putDirect(vm, Identifier::fromString(vm, "address"_s), jsNumber(reinterpret_cast<uintptr_t>(tail.iov_base)));
+        object->putDirect(vm, Identifier::fromString(vm, "length"_s), jsNumber(tail.iov_len));
+        return object;
+    };
+    JSObject* result = constructEmptyObject(globalObject);
+    result->putDirect(vm, Identifier::fromString(vm, "recv"_s), toJS(tails[0]));
+    result->putDirect(vm, Identifier::fromString(vm, "tls"_s), toJS(tails[1]));
+    return JSValue::encode(result);
+#else
+    return encodedJSUndefined();
+#endif
+}
+
+JSC_DEFINE_HOST_FUNCTION(jsFunction_residentBytes, (JSC::JSGlobalObject * globalObject, JSC::CallFrame* callFrame))
+{
+#if OS(LINUX)
+    // test/aot/builtins/native-contract.test.ts calls every native function with operands of every kind.
+    constexpr double maxLength = 16 * 1024 * 1024;
+    JSValue addressValue = callFrame->argument(0);
+    JSValue lengthValue = callFrame->argument(1);
+    if (!addressValue.isNumber() || !lengthValue.isNumber())
+        return encodedJSUndefined();
+    if (!(addressValue.asNumber() >= 0 && addressValue.asNumber() <= JSC::maxSafeInteger()) || !(lengthValue.asNumber() >= 0 && lengthValue.asNumber() <= maxLength))
+        return JSValue::encode(jsNumber(-EINVAL));
+    auto* address = reinterpret_cast<void*>(static_cast<uintptr_t>(addressValue.asNumber()));
+    size_t length = static_cast<size_t>(lengthValue.asNumber());
+    size_t pageSize = static_cast<size_t>(sysconf(_SC_PAGESIZE));
+    Vector<unsigned char> pages(WTF::FillWith {}, (length + pageSize - 1) / pageSize, 0);
+    if (length && mincore(address, length, pages.mutableSpan().data()) != 0)
+        return JSValue::encode(jsNumber(-errno));
+    size_t resident = 0;
+    for (unsigned char page : pages)
+        resident += page & 1;
+    return JSValue::encode(jsNumber(resident * pageSize));
+#else
+    return encodedJSUndefined();
+#endif
+}
+
 #if !OS(WINDOWS)
 static void* spawnThreadsForTestingEntry(void* arg)
 {