diff --git a/data/loongarch_musl/etc/init.d/rcS b/data/loongarch_musl/etc/init.d/rcS index 5a93f5f9..08fb4ed6 100755 --- a/data/loongarch_musl/etc/init.d/rcS +++ b/data/loongarch_musl/etc/init.d/rcS @@ -118,7 +118,7 @@ stage_musl_group_to_tmpfs() { items="lua lua_testcode.sh test.sh date.lua file_io.lua max_min.lua random.lua remove.lua round_num.lua sin30.lua sort.lua strings.lua" ;; iperf_testcode.sh) - items="iperf3 iperf_testcode.sh" + items="busybox iperf3 iperf_testcode.sh" ;; *) return 1 diff --git a/data/risc-v_musl/etc/init.d/rcS b/data/risc-v_musl/etc/init.d/rcS index 5a93f5f9..e9053667 100755 --- a/data/risc-v_musl/etc/init.d/rcS +++ b/data/risc-v_musl/etc/init.d/rcS @@ -95,70 +95,34 @@ mount_official_test_image_if_present() { return 1 } -stage_musl_group_to_tmpfs() { +stage_all_musl_to_tmpfs() { src="/tests/musl" dst="/tmp/musl" - script="$1" [ -d "$src" ] || return 1 - [ -n "$script" ] || return 1 [ -d "$dst" ] || mkdir "$dst" || return 1 - echo "[Tests] staging $script into tmpfs at $dst" - - case "$script" in - basic_testcode.sh) - items="basic basic_testcode.sh busybox" - ;; - busybox_testcode.sh) - items="busybox busybox_cmd.txt busybox_testcode.sh" - ;; - lua_testcode.sh) - items="lua lua_testcode.sh test.sh date.lua file_io.lua max_min.lua random.lua remove.lua round_num.lua sin30.lua sort.lua strings.lua" - ;; - iperf_testcode.sh) - items="iperf3 iperf_testcode.sh" - ;; - *) - return 1 - ;; - esac - - for item in $items; do - [ -e "$src/$item" ] || continue - [ -e "$dst/$item" ] && continue - cp -R "$src/$item" "$dst/" || return 1 - done - - [ -f "$dst/$script" ] || return 1 - echo "[Tests] staged $script into tmpfs" + echo "[Tests] staging full musl test tree into tmpfs at $dst" + cp -R "$src"/. "$dst/" || return 1 + echo "[Tests] staged full musl test tree into tmpfs" return 0 } run_musl_tests_if_present() { - # 官方测试镜像挂载到 /tests 后，当前自动入口只跑 musl 分组； + # 官方测试镜像挂载到 /tests 后，自动扫描并运行 musl 下的全部测试脚本； # glibc 分组保留在测试镜像中，后续需要时直接调整本脚本。 mount_official_test_image_if_present || true - [ -d /tests ] || return 1 + [ -d /tests/musl ] || return 1 ran=0 - echo "[Tests] detected whitelisted musl test scripts; running in rcS" - export PATH="/bin:/sbin:/usr/bin:/usr/sbin:/tmp/musl:/tests/musl" + echo "[Tests] detected musl test scripts; running all in rcS" + stage_all_musl_to_tmpfs || return 1 + + export PATH="/bin:/sbin:/usr/bin:/usr/sbin:/tmp/musl" export HOME="/" - for name in \ - basic_testcode.sh \ - busybox_testcode.sh \ - lua_testcode.sh \ - iperf_testcode.sh - do - if stage_musl_group_to_tmpfs "$name"; then - f="/tmp/musl/$name" - else - echo "[Tests] tmpfs staging failed for $name; running directly from /tests" - f="/tests/musl/$name" - fi + for f in /tmp/musl/*_testcode.sh; do [ -f "$f" ] || continue dir="${f%/*}" diff --git a/docs/os-myself/big-bug-record.md b/docs/os-myself/big-bug-record.md index 11dd6bfc..7ae4de46 100644 --- a/docs/os-myself/big-bug-record.md +++ b/docs/os-myself/big-bug-record.md @@ -74,3 +74,100 @@ BusyBox applet 的链接森林只是在运行时需要，不应该作为 Git 仓 2. 减少 `mount`/`umount` 测试里的 VFAT 初始化成本。 3. 对目录项查找和路径解析加缓存，减少 exec 高频路径的重复 ext4 访问。 4. 如果评测机总时间更长，可以保留当前按组懒 staging；如果只追 basic 分数，可以进一步只运行 basic 组，避免后续组影响关机和输出。 + +## 2026-06-23 iperf TCP 吞吐从几 KB/s 提升到数百 Mbit/s + +### 现象 + +最初 iperf 组能启动，但 TCP 分数几乎没有贡献： + +- `BASIC_TCP` 只发送 4KiB，sender 约 16Kbit/s，receiver 为 0。 +- `REVERSE_TCP` sender 也只有 4KiB，receiver 为 0。 +- `PARALLEL_TCP` 第二条以后连接报 `Connection refused`。 +- UDP 能跑通，说明基本 socket、loopback 地址和 iperf 控制连接不是完全坏掉；问题集中在 TCP 数据面和监听队列语义。 + +### 关键发现 + +第一层瓶颈是监听语义。iperf3 `-P 5` 会为同一个 server port 建多条 TCP data stream。smoltcp 的 TCP socket 没有 Linux 那种单 socket backlog 队列模型，监听端需要准备多个 TCP socket 才能同时接住多个 SYN。原实现只有一个 listener handle，所以第一条连接建立后，后续 stream 容易被拒绝。 + +第二层瓶颈是 loopback MTU 和 adapter RX buffer 不一致。loopback 设备暴露 `mtu=65535`，smoltcp 因此会发接近 64KiB 的 TCP/IP 包；但 `NetDeviceAdapter` 原来只有 2048 字节接收缓冲。结果大 TCP frame 在 adapter 层直接被丢掉，表现为客户端以为发出了少量数据，服务端几乎收不到。 + +第三层是 syscall 和轮询节奏。iperf3 会传大 buffer；如果每次 syscall 都按用户长度一次性分配/copy，会增加内核堆压力，也让 loopback 队列 drain 不及时。写入后不主动 poll，会把推进 TCP 状态机的工作推迟到后续调度点，吞吐很差。 + +### 处理步骤 + +这轮 TCP 相关修复按小步提交推进： + +- `da0d56c tests: stage busybox for iperf script`：iperf 脚本里会调用 BusyBox 工具，staging 只复制 `iperf3` 不够，先保证测试脚本自身能稳定运行。 +- `56c01e8 net: enlarge socket buffers for iperf`：TCP/UDP buffer 扩到 256KiB，避免 iperf 数据面频繁因为小 buffer 进入 WouldBlock。 +- `de092e8 net: chunk socket syscall buffers`：`send/recv` 单次内核复制限制到 64KiB，控制临时分配和 copy 成本。 +- `e63c8a5 net: drain loopback after socket writes`：socket write/sendto 成功后 bounded poll，尽快把 loopback Tx frame 回灌到 Rx 并推进 smoltcp 状态机。 +- `eb9d8ce net: support parallel tcp listeners`：为监听 socket 维护一组 spare TCP listener，accept 时把已建立连接交给新 fd，并补充新的 listener，解决 `PARALLEL_TCP` 连接拒绝。 +- `491634c net: reduce iperf socket log noise`：把热路径 socket/TCP 连接日志从 info 降到 debug，避免串口日志本身拖慢性能。 +- `aae6897 net: size adapter rx buffer for mtu`：adapter RX buffer 改为按 `device.mtu() + Ethernet header` 分配，修复 64KiB loopback TCP frame 被 2048B buffer 丢弃的问题。 +- `474b6ea net: cap tcp buffers below unstable window`：验证 UDP 改动时复现 smoltcp `SeqNumber` subtraction underflow，最终把 TCP buffer 从 256KiB 收到 128KiB-1，避开不稳定的大窗口组合。 + +### 决策和取舍 + +没有直接 patch smoltcp。这里的问题主要是我们给 smoltcp 的设备能力、buffer 和 listen 模型不一致，先修本内核适配层更稳，也更容易解释。 + +TCP buffer 起初选 256KiB，是为了减少 WouldBlock 并提高窗口；这一版峰值很好，`BASIC_TCP`/`REVERSE_TCP` 能到约 490Mbit/s，`PARALLEL_TCP` 能到约 709Mbit/s。但后续在完整 iperf-only 复测中，`PARALLEL_TCP` 五条连接建立后触发 smoltcp sequence number underflow panic。64KiB 以内能稳定，但单流 TCP 只剩约 70Mbit/s。最终选 `128KiB-1`：仍然明显高于原始 4KiB/0 receiver，且完整 iperf 组稳定通过。 + +listener pool 做了上限，不直接相信 iperf 传入的巨大 backlog。当前逻辑把用户 backlog clamp 到 128，实际 spare listener pool clamp 到 16，足够覆盖 iperf `-P 5`，同时避免一次 listen 分配过多 TCP buffer。 + +保留 loopback 的大 MTU，而不是降回 1500。降 MTU 能减少单帧内存，但 TCP throughput 会被更多包处理开销限制。真正的问题是 adapter 宣告大 MTU 却没有同等大小的 RX buffer，因此修 buffer 更符合设备模型。 + +### 验证 + +临时把 RISC-V `rcS` 改成只跑 `iperf_testcode.sh`，构建 release 内核，用临时分区盘 `/tmp/ccyos-disk-udpq-tcp128k.img` 跑官方 RISC-V musl iperf 组。 + +最终采用 128KiB-1 TCP buffer 后，最新一次验证中 TCP 三项都通过： + +- `BASIC_TCP`：sender 71.7MiB / 300Mbit/s，receiver 71.4MiB / 299Mbit/s。 +- `PARALLEL_TCP`：5 stream 全部连接成功，SUM sender 102MiB / 424Mbit/s，receiver 102MiB / 418Mbit/s。 +- `REVERSE_TCP`：sender 72.2MiB / 302Mbit/s，receiver 72.0MiB / 302Mbit/s。 + +这说明最初 “4KiB / receiver 0” 和 “parallel connection refused” 两个 TCP 主问题已经解决。TCP 的峰值为了稳定性放弃了 256KiB buffer 下的最高数字，但所有 TCP 子项稳定通过，且吞吐仍是原始结果的数量级提升。 + +## 2026-06-23 iperf UDP 单流和反向模式高丢包 + +### 现象 + +TCP 主问题解决后，UDP 仍然拖分： + +- `BASIC_UDP` receiver 约 46Mbit/s，丢包约 58%-61%。 +- `REVERSE_UDP` receiver 约 48Mbit/s，同样有大量丢包。 +- `PARALLEL_UDP` 反而能到约 137Mbit/s 且 0% loss。 + +这个现象说明链路、loopback MTU 和基本 UDP socket 不是完全坏掉；单 fd 收包路径比多 stream 分摊路径更容易溢出。 + +### 关键发现 + +本内核对 UDP 做了 per-port smoltcp socket 加 per-fd 队列的分发。smoltcp 共享 socket 收到包以后，`udp_dispatch_drain_locked()` 会复制到目标 `SocketFile` 的 `udp_rx_queue`，用户态再从这个队列 `recvfrom()`。 + +原来的 `UDP_RXQ_CAP = 64` 太小。iperf UDP 使用极高目标带宽发包，发送端在短时间内产生的 datagram 比接收线程 drain 得更快；队列满后 `udp_push()` 直接返回 false，后续 datagram 被丢掉。`PARALLEL_UDP` 因为 5 条 stream 分散到多个 fd 队列，每个队列压力小，所以表现比单流好。 + +试过把 UDP 队列固定预分配到 512。这个版本把 `BASIC_UDP` 提到 77.8Mbit/s 且 0% loss，但 `PARALLEL_UDP` 随后触发 `memory allocation of 1064960 bytes failed`。原因是每个 UDP 队列项包含最大 2048 字节 payload 缓冲，512 项接近 1MiB；所有 UDP socket 一创建就预分配，会把 parallel 场景的内存压力放大。 + +### 处理步骤 + +- `f4905d9 net: grow udp receive queues lazily`：UDP per-fd 队列初始仍为 64，只有队列满时才尝试翻倍扩容，最大 512。 +- 如果扩容失败或已达上限，队列丢弃最旧 datagram 再放入新 datagram。这样在内存紧张时仍然有 UDP 丢包，但不会因为固定大队列把内核堆打爆，也更偏向保留较新的 iperf 数据。 + +### 决策和取舍 + +没有把所有 UDP socket 都预分配 512 项。固定大队列能改善单流 loss，但并行时内存占用不可控；lazy growth 只让真正承压的 socket 付出内存成本。 + +没有把 UDP 队列做成无限增长。iperf `-u -b 1000G` 本质上会制造远超内核处理能力的 burst；无限增长只是把丢包变成内存耗尽。512 是这次本地验证中能消掉单流丢包、又不会让 parallel UDP 分配失败的上限。 + +丢弃策略从“队列满就丢新包”变成“扩不动时丢旧包”。这对 UDP 是可接受取舍：UDP 本来不保证可靠性，保留较新的 datagram 对 iperf 的实时统计和反向模式更有价值。 + +### 验证 + +临时把 RISC-V `rcS` 改成只跑 `iperf_testcode.sh`，构建 release 内核并用临时分区盘 `/tmp/ccyos-disk-udpq-tcp128k.img` 跑官方 RISC-V musl iperf 组。最新一次完整 iperf-only 验证全部通过： + +- `BASIC_UDP`：sender 25.9MiB / 109Mbit/s，receiver 25.9MiB / 108Mbit/s，0% loss。 +- `PARALLEL_UDP`：SUM sender 36.6MiB / 154Mbit/s，receiver 36.5MiB / 153Mbit/s，0% loss。 +- `REVERSE_UDP`：sender 19.9MiB / 82.7Mbit/s，receiver 19.8MiB / 82.8Mbit/s，0% loss。 + +这说明 UDP 的主要问题是 per-fd 接收队列容量和内存策略，而不是 UDP 校验、地址绑定或 loopback 设备本身。 diff --git a/os/src/kernel/syscall/network/connection_ops.rs b/os/src/kernel/syscall/network/connection_ops.rs index 20ac4788..6b203654 100644 --- a/os/src/kernel/syscall/network/connection_ops.rs +++ b/os/src/kernel/syscall/network/connection_ops.rs @@ -1,5 +1,7 @@ use super::*; +const SOCKET_IO_CHUNK_SIZE: usize = 64 * 1024; + /// 连接到远程地址 pub fn connect(sockfd: i32, addr: *const u8, addrlen: u32) -> isize { let task = current_task(); @@ -140,7 +142,7 @@ pub fn connect(sockfd: i32, addr: *const u8, addrlen: u32) -> isize { } pr_debug!("connect: tcp_connect success, nonblock={}", is_nonblock); - crate::pr_info!( + crate::pr_debug!( "[TCP] Connection established: {} -> {}", local_endpoint, endpoint @@ -233,13 +235,14 @@ pub fn send(sockfd: i32, buf: *const u8, len: usize, _flags: i32) -> isize { (tid, file) }; + let chunk_len = core::cmp::min(len, SOCKET_IO_CHUNK_SIZE); let result = { - let mut kernel_buf = alloc::vec![0u8; len]; + let mut kernel_buf = alloc::vec![0u8; chunk_len]; unsafe { crate::arch::ArchImpl::copy_from_user( crate::arch::address::UA::from_usize(buf as usize), kernel_buf.as_mut_ptr(), - len, + chunk_len, ) .ok(); } @@ -248,11 +251,11 @@ pub fn send(sockfd: i32, buf: *const u8, len: usize, _flags: i32) -> isize { match result { Ok(n) => { - pr_debug!("send: sockfd={}, len={} -> sent={}", sockfd, len, n); + pr_debug!("send: sockfd={}, len={} -> sent={}", sockfd, chunk_len, n); return n as isize; } Err(e) => { - pr_debug!("send: sockfd={}, len={} -> error={:?}", sockfd, len, e); + pr_debug!("send: sockfd={}, len={} -> error={:?}", sockfd, chunk_len, e); if e == crate::vfs::FsError::WouldBlock { if let Some(socket_file) = file.as_any().downcast_ref::() && !socket_file.flags().contains(OpenFlags::O_NONBLOCK) @@ -301,8 +304,9 @@ pub fn recv(sockfd: i32, buf: *mut u8, len: usize, _flags: i32) -> isize { (tid, file) }; + let chunk_len = core::cmp::min(len, SOCKET_IO_CHUNK_SIZE); let result = { - let mut kernel_buf = alloc::vec![0u8; len]; + let mut kernel_buf = alloc::vec![0u8; chunk_len]; match file.read(&mut kernel_buf) { Ok(n) => { unsafe { @@ -321,11 +325,11 @@ pub fn recv(sockfd: i32, buf: *mut u8, len: usize, _flags: i32) -> isize { match result { Ok(n) => { - pr_debug!("recv: sockfd={}, len={} -> received={}", sockfd, len, n); + pr_debug!("recv: sockfd={}, len={} -> received={}", sockfd, chunk_len, n); return n as isize; } Err(e) => { - pr_debug!("recv: sockfd={}, len={} -> error={:?}", sockfd, len, e); + pr_debug!("recv: sockfd={}, len={} -> error={:?}", sockfd, chunk_len, e); if e == crate::vfs::FsError::WouldBlock { if let Some(socket_file) = file.as_any().downcast_ref::() && !socket_file.flags().contains(OpenFlags::O_NONBLOCK) diff --git a/os/src/kernel/syscall/network/mod.rs b/os/src/kernel/syscall/network/mod.rs index 3582be90..a95ad98f 100644 --- a/os/src/kernel/syscall/network/mod.rs +++ b/os/src/kernel/syscall/network/mod.rs @@ -112,7 +112,7 @@ use crate::{ wait_unix_would_block, write_sockaddr_un, write_socketpair_fds, }, }, - pr_debug, pr_info, println, + pr_debug, println, uapi::{ fcntl::{FdFlags, OpenFlags}, socket::{AF_UNIX, SOCK_CLOEXEC, SOCK_DGRAM, SOCK_NONBLOCK, SOCK_STREAM, SOCK_TYPE_MASK}, diff --git a/os/src/kernel/syscall/network/socket_ops.rs b/os/src/kernel/syscall/network/socket_ops.rs index 8642f13e..a5371700 100644 --- a/os/src/kernel/syscall/network/socket_ops.rs +++ b/os/src/kernel/syscall/network/socket_ops.rs @@ -1,5 +1,7 @@ use super::*; +const TCP_LISTENER_POOL_LIMIT: usize = 16; + /// 创建套接字 pub fn socket(domain: i32, socket_type: i32, _protocol: i32) -> isize { let base_type = socket_type & SOCK_TYPE_MASK; @@ -61,7 +63,7 @@ pub fn socket(domain: i32, socket_type: i32, _protocol: i32) -> isize { SocketHandle::Tcp(_) => "TCP", SocketHandle::Udp(_) => "UDP", }; - pr_info!( + pr_debug!( "[SOCKET] Created {} socket: tid={}, fd={}, domain={}, type={}", handle_type, tid, @@ -290,7 +292,7 @@ pub fn listen(sockfd: i32, backlog: i32) -> isize { ); let mut attempts_left: usize = if endpoint.port == 0 { 32 } else { 1 }; - loop { + let listen_endpoint = loop { if endpoint.port == 0 { endpoint.port = alloc_ephemeral_port(); } @@ -301,7 +303,7 @@ pub fn listen(sockfd: i32, backlog: i32) -> isize { // Convert endpoint to listen endpoint // If bound to 0.0.0.0 or ::, listen on all addresses (addr = None) - use smoltcp::wire::{IpAddress, IpListenEndpoint}; + use smoltcp::wire::IpListenEndpoint; let listen_endpoint = match endpoint.addr { IpAddress::Ipv4(addr) if addr.is_unspecified() => IpListenEndpoint { addr: None, @@ -332,14 +334,17 @@ pub fn listen(sockfd: i32, backlog: i32) -> isize { endpoint.port = 0; continue; } - break; - } + break listen_endpoint; + }; socket_file.set_listener(true); socket_file.clear_listen_sockets(); // iperf 会传入非常大的 backlog（甚至 INT_MAX），这里做一个上限避免内存/逻辑风险 let backlog = (backlog as usize).clamp(1, 128); socket_file.set_listen_backlog(backlog); + if let Err(e) = replenish_tcp_listeners(socket_file, listen_endpoint, backlog) { + return e; + } 0 } SocketHandle::Udp(_) => { @@ -426,19 +431,17 @@ pub fn accept(sockfd: i32, addr: *mut u8, addrlen: *mut u32) -> isize { None => return -88, // ENOTSOCK }; - if state != TcpListenState::Listen && socket_file.listen_sockets_len() < backlog { - // detach current listen socket immediately - let new_listen_handle = match create_tcp_socket() { - Ok(SocketHandle::Tcp(h)) => h, - Err(e) => return e.to_errno(), - Ok(SocketHandle::Udp(_)) => return -(crate::uapi::errno::EINVAL as isize), - }; - - if let Err(e) = network_stack().tcp_listen(new_listen_handle, listen_endpoint) { - network_stack().remove_tcp_socket(new_listen_handle); - return e.to_errno(); + if state != TcpListenState::Listen { + if let Err(e) = replenish_tcp_listeners(socket_file, listen_endpoint, backlog) { + return e; } + let Some(new_listen_handle) = + network_stack().take_spare_tcp_listener(socket_file, listen_endpoint) + else { + return -11; // EAGAIN + }; + use crate::net::socket::{update_socket_file_handle, update_socket_handle}; update_socket_handle( tid as usize, @@ -464,10 +467,19 @@ pub fn accept(sockfd: i32, addr: *mut u8, addrlen: *mut u32) -> isize { } // Otherwise: keep it as pending (SynReceived, etc). - socket_file.add_listen_socket(SocketHandle::Tcp(listen_handle)); + if !socket_file.has_listen_socket(SocketHandle::Tcp(listen_handle)) { + socket_file.add_listen_socket(SocketHandle::Tcp(listen_handle)); + } + if let Err(e) = replenish_tcp_listeners(socket_file, listen_endpoint, backlog) { + return e; + } continue; } + if let Err(e) = replenish_tcp_listeners(socket_file, listen_endpoint, backlog) { + return e; + } + if is_nonblock { return -11; // EAGAIN } @@ -478,6 +490,29 @@ pub fn accept(sockfd: i32, addr: *mut u8, addrlen: *mut u32) -> isize { } } +fn replenish_tcp_listeners( + socket_file: &SocketFile, + listen_endpoint: smoltcp::wire::IpListenEndpoint, + backlog: usize, +) -> Result<(), isize> { + let target = backlog.clamp(1, TCP_LISTENER_POOL_LIMIT); + while network_stack().tcp_spare_listener_count(socket_file, listen_endpoint) < target { + let new_listen_handle = match create_tcp_socket() { + Ok(SocketHandle::Tcp(h)) => h, + Err(e) => return Err(e.to_errno()), + Ok(SocketHandle::Udp(_)) => return Err(-(crate::uapi::errno::EINVAL as isize)), + }; + + if let Err(e) = network_stack().tcp_listen(new_listen_handle, listen_endpoint) { + network_stack().remove_tcp_socket(new_listen_handle); + return Err(e.to_errno()); + } + + socket_file.add_listen_socket(SocketHandle::Tcp(new_listen_handle)); + } + Ok(()) +} + fn accept_return_conn( task: crate::kernel::SharedTask, tid: usize, diff --git a/os/src/net/socket.rs b/os/src/net/socket.rs index 57fb5693..81577ce6 100644 --- a/os/src/net/socket.rs +++ b/os/src/net/socket.rs @@ -26,7 +26,8 @@ lazy_static! { use crate::uapi::fcntl::OpenFlags; use crate::uapi::socket::SocketOptions; -const UDP_RXQ_CAP: usize = 64; +const UDP_RXQ_INITIAL_CAP: usize = 64; +const UDP_RXQ_MAX_CAP: usize = 512; pub(crate) const UDP_DGRAM_MAX: usize = 2048; #[derive(Debug)] @@ -58,7 +59,7 @@ impl SocketFile { listen_backlog: SpinLock::new(0), local_endpoint: SpinLock::new(None), remote_endpoint: SpinLock::new(None), - udp_rx_queue: SpinLock::new(VecDeque::with_capacity(UDP_RXQ_CAP)), + udp_rx_queue: SpinLock::new(VecDeque::with_capacity(UDP_RXQ_INITIAL_CAP)), shutdown_rd: SpinLock::new(false), shutdown_wr: SpinLock::new(false), flags: SpinLock::new(OpenFlags::empty()), @@ -82,7 +83,7 @@ impl SocketFile { listen_backlog: SpinLock::new(0), local_endpoint: SpinLock::new(None), remote_endpoint: SpinLock::new(None), - udp_rx_queue: SpinLock::new(VecDeque::with_capacity(UDP_RXQ_CAP)), + udp_rx_queue: SpinLock::new(VecDeque::with_capacity(UDP_RXQ_INITIAL_CAP)), shutdown_rd: SpinLock::new(false), shutdown_wr: SpinLock::new(false), flags: SpinLock::new(flags), @@ -123,8 +124,15 @@ impl SocketFile { self.listen_sockets.lock().clear(); } - pub fn listen_sockets_len(&self) -> usize { - self.listen_sockets.lock().len() + pub fn has_listen_socket(&self, handle: SocketHandle) -> bool { + self.listen_sockets + .lock() + .iter() + .any(|h| match (*h, handle) { + (SocketHandle::Tcp(a), SocketHandle::Tcp(b)) => a == b, + (SocketHandle::Udp(a), SocketHandle::Udp(b)) => a == b, + _ => false, + }) } /// Pop one established connection from the listener queue. @@ -162,7 +170,16 @@ impl SocketFile { pub(crate) fn udp_push(&self, d: UdpDatagram) -> bool { let mut q = self.udp_rx_queue.lock(); if q.len() == q.capacity() { - return false; + let old_capacity = q.capacity(); + if old_capacity < UDP_RXQ_MAX_CAP { + let new_capacity = old_capacity + .saturating_mul(2) + .clamp(UDP_RXQ_INITIAL_CAP, UDP_RXQ_MAX_CAP); + let _ = q.try_reserve(new_capacity.saturating_sub(old_capacity)); + } + if q.len() == q.capacity() { + let _ = q.pop_front(); + } } q.push_back(d); true diff --git a/os/src/net/stack/adapter.rs b/os/src/net/stack/adapter.rs index f9c5c1c9..a497abf1 100644 --- a/os/src/net/stack/adapter.rs +++ b/os/src/net/stack/adapter.rs @@ -63,15 +63,19 @@ impl SmoltcpInterface { #[derive(Clone)] pub struct NetDeviceAdapter { device: Arc, - rx_buffer: [u8; 2048], + rx_buffer: Vec, } impl NetDeviceAdapter { /// Create a new adapter. pub fn new(device: Arc) -> Self { + let rx_len = device + .mtu() + .saturating_add(smoltcp::wire::EthernetFrame::<&[u8]>::header_len()) + .max(2048); Self { device, - rx_buffer: [0; 2048], + rx_buffer: alloc::vec![0; rx_len], } } diff --git a/os/src/net/stack/mod.rs b/os/src/net/stack/mod.rs index 87a2c3b8..eaf4134e 100644 --- a/os/src/net/stack/mod.rs +++ b/os/src/net/stack/mod.rs @@ -18,6 +18,16 @@ use super::socket::{self, SocketFile, SocketHandle, UdpDatagram}; mod adapter; pub use adapter::{NetDeviceAdapter, SmoltcpInterface}; +// 256KiB buffers can drive smoltcp's large-window path into a sequence underflow +// under parallel loopback iperf; 128KiB-1 keeps throughput high with a stable window scale. +const TCP_RX_BUFFER_SIZE: usize = 128 * 1024 - 1; +const TCP_TX_BUFFER_SIZE: usize = 128 * 1024 - 1; +const UDP_PACKET_METADATA_CAPACITY: usize = 256; +const UDP_RX_BUFFER_SIZE: usize = 256 * 1024; +const UDP_TX_BUFFER_SIZE: usize = 256 * 1024; +const LOOPBACK_WRITE_DRAIN_POLLS: usize = 64; +const LOOPBACK_FULL_DRAIN_POLLS: usize = 256; + pub(crate) fn enqueue_loopback_frame(frame: Vec) { network_stack().enqueue_loopback_frame(frame); } @@ -154,15 +164,15 @@ impl NetworkStack { pub fn create_tcp_socket(&self) -> Result { let mut rx_vec = alloc::vec::Vec::new(); rx_vec - .try_reserve(4096) + .try_reserve(TCP_RX_BUFFER_SIZE) .map_err(|_| NetworkError::NoMemory)?; - rx_vec.resize(4096, 0); + rx_vec.resize(TCP_RX_BUFFER_SIZE, 0); let mut tx_vec = alloc::vec::Vec::new(); tx_vec - .try_reserve(4096) + .try_reserve(TCP_TX_BUFFER_SIZE) .map_err(|_| NetworkError::NoMemory)?; - tx_vec.resize(4096, 0); + tx_vec.resize(TCP_TX_BUFFER_SIZE, 0); let rx_buffer = tcp::SocketBuffer::new(rx_vec); let tx_buffer = tcp::SocketBuffer::new(tx_vec); @@ -412,6 +422,10 @@ impl NetworkStack { /// Poll until loopback compatibility queues are boundedly drained. pub fn poll_until_empty(&self) { + self.poll_loopback_bounded(LOOPBACK_FULL_DRAIN_POLLS); + } + + fn poll_loopback_bounded(&self, max_polls: usize) { if let Some(ref wrapper) = *self.net_iface.lock() { wrapper.poll_smoltcp(&self.socket_set); { @@ -420,8 +434,7 @@ impl NetworkStack { } self.reap_pending_tcp_close(); - const MAX_DRAIN_POLLS: usize = 256; - for _ in 0..MAX_DRAIN_POLLS { + for _ in 0..max_polls { if wrapper.loopback_queue_len() == 0 { break; } @@ -465,6 +478,76 @@ impl NetworkStack { None } + /// Pop one spare listening TCP socket from a listener pool. + pub fn take_spare_tcp_listener( + &self, + file: &super::socket::SocketFile, + listen_endpoint: IpListenEndpoint, + ) -> Option { + let sockets = self.socket_set.lock(); + let mut q = file.listen_sockets.lock(); + let mut i = 0; + while i < q.len() { + match q[i] { + SocketHandle::Tcp(h) => { + let s = sockets.get::(h); + match s.state() { + tcp::State::Listen if s.listen_endpoint() == listen_endpoint => { + q.remove(i); + return Some(h); + } + tcp::State::Closed => { + q.remove(i); + continue; + } + _ => {} + } + } + SocketHandle::Udp(_) => { + q.remove(i); + continue; + } + } + i += 1; + } + None + } + + /// Count spare listening sockets for an endpoint, pruning stale queue entries. + pub fn tcp_spare_listener_count( + &self, + file: &super::socket::SocketFile, + listen_endpoint: IpListenEndpoint, + ) -> usize { + let sockets = self.socket_set.lock(); + let mut q = file.listen_sockets.lock(); + let mut count = 0; + let mut i = 0; + while i < q.len() { + match q[i] { + SocketHandle::Tcp(h) => { + let s = sockets.get::(h); + match s.state() { + tcp::State::Listen if s.listen_endpoint() == listen_endpoint => { + count += 1; + } + tcp::State::Closed => { + q.remove(i); + continue; + } + _ => {} + } + } + SocketHandle::Udp(_) => { + q.remove(i); + continue; + } + } + i += 1; + } + count + } + /// Query whether a socket has reached the closed TCP state. pub fn socket_is_closed(&self, file: &super::socket::SocketFile) -> bool { let sockets = self.socket_set.lock(); @@ -540,7 +623,7 @@ impl NetworkStack { } }; if result.is_ok() { - self.poll(); + self.poll_loopback_bounded(LOOPBACK_WRITE_DRAIN_POLLS); crate::kernel::syscall::io::wake_poll_waiters(); } result @@ -736,7 +819,7 @@ impl NetworkStack { } }; if result.is_ok() { - self.poll(); + self.poll_loopback_bounded(LOOPBACK_WRITE_DRAIN_POLLS); crate::kernel::syscall::io::wake_poll_waiters(); } result @@ -806,27 +889,27 @@ impl NetworkStack { ) -> Result { let mut rx_meta_vec = alloc::vec::Vec::new(); rx_meta_vec - .try_reserve(4) + .try_reserve(UDP_PACKET_METADATA_CAPACITY) .map_err(|_| NetworkError::NoMemory)?; - rx_meta_vec.resize(4, udp::PacketMetadata::EMPTY); + rx_meta_vec.resize(UDP_PACKET_METADATA_CAPACITY, udp::PacketMetadata::EMPTY); let mut tx_meta_vec = alloc::vec::Vec::new(); tx_meta_vec - .try_reserve(4) + .try_reserve(UDP_PACKET_METADATA_CAPACITY) .map_err(|_| NetworkError::NoMemory)?; - tx_meta_vec.resize(4, udp::PacketMetadata::EMPTY); + tx_meta_vec.resize(UDP_PACKET_METADATA_CAPACITY, udp::PacketMetadata::EMPTY); let mut rx_data_vec = alloc::vec::Vec::new(); rx_data_vec - .try_reserve(4096) + .try_reserve(UDP_RX_BUFFER_SIZE) .map_err(|_| NetworkError::NoMemory)?; - rx_data_vec.resize(4096, 0); + rx_data_vec.resize(UDP_RX_BUFFER_SIZE, 0); let mut tx_data_vec = alloc::vec::Vec::new(); tx_data_vec - .try_reserve(4096) + .try_reserve(UDP_TX_BUFFER_SIZE) .map_err(|_| NetworkError::NoMemory)?; - tx_data_vec.resize(4096, 0); + tx_data_vec.resize(UDP_TX_BUFFER_SIZE, 0); let rx_buffer = udp::PacketBuffer::new(rx_meta_vec, rx_data_vec); let tx_buffer = udp::PacketBuffer::new(tx_meta_vec, tx_data_vec);