From 0a23ca58e253de31f7753e85a9698e0d50f54d67 Mon Sep 17 00:00:00 2001 From: ningding Date: Mon, 20 Jul 2026 19:19:21 -0700 Subject: [PATCH 1/6] Add multi-waiter support for TCP accept Allow a single thread to wait for incoming connections across multiple listening sockets at once using multiwaiter_wait(), instead of having to block in network_socket_accept_tcp() on one socket at a time. Each socket now exposes an array of futex called eventFutexState, that a caller can register them with a multiwaiter. For this PR, only the added the accept functionality, others remains commented out for now. A thread can watch the accept futexes of several listening sockets simultaneously, block in multiwaiter_wait() until one of them signals a ready connection, and then accept that connection. The blocking wait therefore moves out of accept() and into the multiwaiter: accept() is called only once a connection is known to be ready, so it no longer has to block, which is what makes it possible to wait on several listening sockets at the same time from a single thread. Key changes: Added Event futex array (tcpip-internal.h, NetAPI.h) for socket wrapper: Only SocketAcceptEvent is implemented today, room reserved for future receive/send events. The value of the futex counts the ready events currently pending on the socket. For SocketAcceptEvent, it represents the number of connections that are ready to be accepted. The reserved value SocketNotAvailable (0xFFFFFFFF) means the socket will be freed soon. Read-only getter: network_socket_get_event_source() returns a capability to a socket's event futex stripped to read-only. Producing events: on_tcp_connect increments the SocketAcceptEvent futex and calls notify_all() when a new connection becomes ready. To let the callback can find the wrapper from the raw socket, network_socket_create_and_bind() now applys a bidirectional link between the wrapper and raw socket. Consuming events: network_socket_accept_tcp() decrements the futex after it successfully accepts a connection. This only keeps the futex value align with the semantic desbribed above. It does not call notify_all(), since dequeuing an existing connection is not a new, so no need to wake the threads up. Avoid waiting forever when socket is not available: the futex lives in the socket wrapper's heap memory, so a thread must never be left blocked on a socket that is being freed. Both network_socket_close() and the reset handler (reset_network_stack_state) set every event futex to SocketNotAvailable and call notify_all() before that memory goes away. The value change wakes any thread blocked in multiwaiter_wait(), which then observes the sentinel and returns instead of sleeping on a dead socket. A subsequent accept() will see the sentinel and reports the socket as unavailable. --- include/NetAPI.h | 27 +++++++ lib/tcpip/network_wrapper.cc | 133 ++++++++++++++++++++++++++++++++ lib/tcpip/tcpip-internal.h | 10 +++ lib/tcpip/tcpip_error_handler.h | 7 ++ 4 files changed, 177 insertions(+) diff --git a/include/NetAPI.h b/include/NetAPI.h index 65be5f0e..0d949a8e 100644 --- a/include/NetAPI.h +++ b/include/NetAPI.h @@ -34,6 +34,25 @@ struct NetworkAddress } kind; }; +/** + * Enumeration that defines futex types. Each value corresponds + * to an index in the socket's futex array. + */ +enum SocketEventType : uint8_t +{ + SocketAcceptEvent = 0, // Triggered when a new TCP connection is accepted on + // a listening socket + // SocketReceiveEvent = 1, // Triggered when data is received on a socket + // SocketSendEvent = 2 // Triggered when a socket has space available + // for sending +}; + +/// Number of distinct socket event futex types. +static constexpr size_t NumFutexTypes = SocketEventType::SocketAcceptEvent + 1; +/// Sentinel value stored in a socket event futex after the socket has been +/// torn down. +static constexpr uint32_t SocketNotAvailable = -1; + /** * Enumeration defining the connection type. */ @@ -269,6 +288,14 @@ Socket __cheri_compartment("TCPIP") AllocatorCapability mallocCapability, bool isIPv6); +/** + * Return the event source associated with a socket. + * + * The returned capability is read-only and bounded to four bytes. + */ +uint32_t *__cheri_compartment("TCPIP") + network_socket_get_event_source(Socket sealedSocket, SocketEventType type); + /** * Authorise a UDP socket to send packets to a specific host. This opens a * firewall hole allowing the socket to send and receive packets to the host. diff --git a/lib/tcpip/network_wrapper.cc b/lib/tcpip/network_wrapper.cc index f8562f51..b96bf267 100644 --- a/lib/tcpip/network_wrapper.cc +++ b/lib/tcpip/network_wrapper.cc @@ -412,8 +412,48 @@ namespace return ret; } + } // namespace +int SealedSocket::signal_event_futex(SocketEventType type) +{ + auto &futex = eventFutexState[type]; + if (heap_claim_ephemeral(TimeoutWaitForever, &futex) != 0) + { + return -EINVAL; + } + uint32_t current = futex.load(); + while (current != SocketNotAvailable) + { + if (futex.compare_exchange_strong(current, current + 1)) + { + futex.notify_all(); + return 0; + } + } + + return -EINVAL; +} + +/** + * The caller must hold socketLock, which prevents the SealedSocket from being + * deallocated while this method accesses the futex. + */ +int SealedSocket::consume_event_futex(SocketEventType type) +{ + auto &futex = eventFutexState[type]; + uint32_t current = futex.load(); + while (current != SocketNotAvailable && current != 0) + { + if (futex.compare_exchange_strong(current, current - 1)) + { + return 0; + } + } + + return (current == SocketNotAvailable) ? -EINVAL : 0; +} + /** * Callback called by FreeRTOS+TCP when a TCP connection is created or * terminated. @@ -468,6 +508,24 @@ static void on_tcp_connect(Socket_t socket, BaseType_t isConnected) address.sin_address.ulIP_IPv4, localPort, address.sin_port); } } + else + { + // Update eventFutexState in the listening socket and wake up the + // threads sleeping on the corresponding futex. + // Use the FreeRTOS getter to get the wrapper capability we stored in + // it. + auto *wrapper = + static_cast(pvSocketGetSocketID(socket)); + if (wrapper != nullptr) + { + /* + * Ignore the return value. If the socket has been invalidated, + * the subsequent accept call will detect that it is unavailable. + */ + wrapper->signal_event_futex(SocketEventType::SocketAcceptEvent); + } + return; + } } F_TCP_UDP_Handler_t onTCPConnectCallback = {on_tcp_connect}; @@ -495,6 +553,11 @@ Socket network_socket_create_and_bind(Timeout *timeout, // Set the socket epoch socketWrapper->socketEpoch = currentSocketEpoch.load(); + // Multi-waiter: initialize the futexes. + for (auto &eventState : socketWrapper->eventFutexState) + { + eventState.store(0); + } const auto Family = isIPv6 ? FREERTOS_AF_INET6 : FREERTOS_AF_INET; Socket_t socket = @@ -512,7 +575,14 @@ Socket network_socket_create_and_bind(Timeout *timeout, token_obj_destroy(mallocCapability, socket_key(), sealedSocket); return -ENOMEM; } + + /* + * Create a bidirectional association between the FreeRTOS socket and + * the sealed socket wrapper. This is used to retrieve the sealed + * socket wrapper from the FreeRTOS socket in the callbacks. + */ socketWrapper->socket = socket; + xSocketSetSocketID(socket, socketWrapper); // Claim the socket so that it counts towards the caller's quota. The // network stack also keeps a claim to it. We will drop this claim on @@ -673,6 +743,24 @@ Socket network_socket_accept_tcp(Timeout *timeout, token_obj_destroy(mallocCapability, socket_key(), sealedSocket); return acceptResult; } + /* + * Note that here we update the futex, but no need to call + * notify_all() on the waiting threads. Because the threads are + * waiting on multiwaiter for arriving connected socket, but here the + * semantics is not having an arrving connection. So there is no need + * to wake the threads up, since they will probably find that there is + * no available sockets and go back to sleep again. + */ + int futexConsumeResult = listeningSocket->consume_event_futex( + SocketEventType::SocketAcceptEvent); + if (futexConsumeResult != 0) + { + // Return -EINVAL, which represents the + // socket will be freed soon. + close_socket_retry(timeout, rawSocket); + token_obj_destroy(mallocCapability, socket_key(), sealedSocket); + return futexConsumeResult; + } socketWrapper->socket = rawSocket; // Claim the socket so that it counts towards the caller's quota. The @@ -831,6 +919,37 @@ Socket network_socket_udp(Timeout *timeout, timeout, mallocCapability, isIPv6, ConnectionTypeUDP); } +/** + * Getter to get the read only capability of a specific futex in + * specific socket. If the futex is invalid, return a untagged nullptr. + */ +uint32_t *network_socket_get_event_source(Socket sealedSocket, + SocketEventType type) +{ + uint32_t *result = nullptr; + if (type < NumFutexTypes) + { + with_sealed_socket( + [&](SealedSocket *socket) { + auto *futex = + reinterpret_cast(&socket->eventFutexState[type]); + + Capability readyOnlyEventSource{futex}; + // Restrict the capability to read-only so the futex can only be + // modified through the callback, not by the caller of this + // helper. + readyOnlyEventSource.bounds() = sizeof(uint32_t); + readyOnlyEventSource.permissions() &= + {Permission::Load, Permission::Global}; + + result = readyOnlyEventSource.get(); + return 0; + }, + sealedSocket); + } + return result; +} + int network_socket_close(Timeout *t, AllocatorCapability mallocCapability, Socket sealedSocket) @@ -893,6 +1012,12 @@ int network_socket_close(Timeout *t, if (socketEpoch == currentSocketEpoch.load()) { bool isTCP = rawSocket->ucProtocol == FREERTOS_IPPROTO_TCP; + + // Set the back pointer (pointing to the + // wrapper) to nullptr, so that we will not + // have a dangling pointer. + xSocketSetSocketID(rawSocket, nullptr); + // Nothing to do if `FreeRTOS_shutdown` // fails: this happens only if the TCP // connection is dead, which is likely to @@ -1040,6 +1165,14 @@ int network_socket_close(Timeout *t, // freed next time we try. return -ETIMEDOUT; } + // Wake any thread waiting on this socket's event futex so + // that it does not sleep forever on memory that is about to + // be freed. + for (auto &eventState : socket->eventFutexState) + { + eventState.store(SocketNotAvailable); + eventState.notify_all(); + } g.release(); socket->socketLock.upgrade_for_destruction(); diff --git a/lib/tcpip/tcpip-internal.h b/lib/tcpip/tcpip-internal.h index 64daa12a..056b8447 100644 --- a/lib/tcpip/tcpip-internal.h +++ b/lib/tcpip/tcpip-internal.h @@ -3,6 +3,8 @@ #pragma once #include +#include +#include #include #include #include @@ -42,6 +44,14 @@ struct SealedSocket * to the current instance of the network stack. */ uint64_t socketEpoch; + /** + * Event waiter source futex array. This supports the multi-waiter + * feature. Different events increment different futexes in the array and + * wake the corresponding waiting threads. + */ + std::atomic eventFutexState[NumFutexTypes]; + int signal_event_futex(SocketEventType type); + int consume_event_futex(SocketEventType type); /** * The lock protecting this socket. */ diff --git a/lib/tcpip/tcpip_error_handler.h b/lib/tcpip/tcpip_error_handler.h index 876bda33..90c31287 100644 --- a/lib/tcpip/tcpip_error_handler.h +++ b/lib/tcpip/tcpip_error_handler.h @@ -195,6 +195,13 @@ extern "C" void reset_network_stack_state(bool isIpThread) DebugErrorHandler::log("Ignoring corrupted socket lock {}.", lock); } + // Notify all threads waiting on the socket's futexes so that they + // do not sleep forever on this old socket. + for (auto &eventState : socket->eventFutexState) + { + eventState.store(SocketNotAvailable); + eventState.notify_all(); + } FreeRTOS_Socket_t *s = socket->socket; if (Capability{s}.is_valid() && From 9be0fd6c26c59cde9d72dc78d85efd09381e45d9 Mon Sep 17 00:00:00 2001 From: nding04 Date: Fri, 31 Jul 2026 02:26:59 +0000 Subject: [PATCH 2/6] Fix comments and helper ownership issues for multi-waiter support - Claim `signal_event_futex` and `consume_event_futex` are owned by SealedSocket - Use `TimeoutWaitForever` instead of timeout{UnlimitedTimeout} --- lib/tcpip/network_wrapper.cc | 10 ++++++---- lib/tcpip/tcpip-internal.h | 22 ++++++++++++++++++++-- 2 files changed, 26 insertions(+), 6 deletions(-) diff --git a/lib/tcpip/network_wrapper.cc b/lib/tcpip/network_wrapper.cc index b96bf267..97553298 100644 --- a/lib/tcpip/network_wrapper.cc +++ b/lib/tcpip/network_wrapper.cc @@ -425,6 +425,12 @@ int SealedSocket::signal_event_futex(SocketEventType type) uint32_t current = futex.load(); while (current != SocketNotAvailable) { + /* + * If the futex's current value equals current, + * store current + 1 and return true. Otherwise + * write the actual current value into current + * (by reference) and return false. + */ if (futex.compare_exchange_strong(current, current + 1)) { futex.notify_all(); @@ -435,10 +441,6 @@ int SealedSocket::signal_event_futex(SocketEventType type) return -EINVAL; } -/** - * The caller must hold socketLock, which prevents the SealedSocket from being - * deallocated while this method accesses the futex. - */ int SealedSocket::consume_event_futex(SocketEventType type) { auto &futex = eventFutexState[type]; diff --git a/lib/tcpip/tcpip-internal.h b/lib/tcpip/tcpip-internal.h index 056b8447..98cfd3c4 100644 --- a/lib/tcpip/tcpip-internal.h +++ b/lib/tcpip/tcpip-internal.h @@ -50,8 +50,26 @@ struct SealedSocket * wake the corresponding waiting threads. */ std::atomic eventFutexState[NumFutexTypes]; - int signal_event_futex(SocketEventType type); - int consume_event_futex(SocketEventType type); + /** + * Increments the futex and notifies all waiters if the futex is still + * valid. + * + * Returns 0 on success, or `-EINVAL` if the futex has been invalidated. + */ + int signal_event_futex(SocketEventType type); + /** + * Consume one pending event by decrementing the futex. + * + * The caller must hold `socketLock`, which prevents the socket from being + * deallocated while the futex is accessed. This operation is only + * bookkeeping: the caller has already consumed the corresponding event, so + * a zero counter is not an error. + * + * Returns 0 if the futex remains valid, regardless of whether the counter + * was decremented. Returns `-EINVAL` if the futex has been invalidated by + * being set to `SocketNotAvailable` because the socket is being torn down. + */ + int consume_event_futex(SocketEventType type); /** * The lock protecting this socket. */ From 07f1d363861e6fea2b7eb1e43700a254e0177ee3 Mon Sep 17 00:00:00 2001 From: nding04 Date: Fri, 31 Jul 2026 04:37:37 +0000 Subject: [PATCH 3/6] TCPIP: fix stale accept futex notifications A TCP child socket can become visible and accept-ready to `FreeRTOS_accept()` before `on_tcp_connect()` increments the accept futex. If a waiter accepts the child while the counter is zero, the old futex consume helper does not decrement it to -1, because -1 represents an invalid futex. The delayed `on_tcp_connect` callback then increments the counter to one, though no connection is pending. This leaves pending multi-waiters with a misleading futex. They will repeatedly treat the socket as ready, call `accept()`, and never block, defeating the purpose of multi-waiter waiting. Use a signed counter and always decrement after a successful accept. An early accept records a temporary negative debt that the callback later repays. Reserve INT32_MIN for an invalid futex so that normal negative debt cannot conflict with the sentinel value. --- include/NetAPI.h | 2 +- lib/tcpip/network_wrapper.cc | 29 ++++++++++++++--------------- lib/tcpip/tcpip-internal.h | 17 +++++++++-------- 3 files changed, 24 insertions(+), 24 deletions(-) diff --git a/include/NetAPI.h b/include/NetAPI.h index 0d949a8e..e6f4f00b 100644 --- a/include/NetAPI.h +++ b/include/NetAPI.h @@ -51,7 +51,7 @@ enum SocketEventType : uint8_t static constexpr size_t NumFutexTypes = SocketEventType::SocketAcceptEvent + 1; /// Sentinel value stored in a socket event futex after the socket has been /// torn down. -static constexpr uint32_t SocketNotAvailable = -1; +static constexpr int32_t SocketNotAvailable = INT32_MIN; /** * Enumeration defining the connection type. diff --git a/lib/tcpip/network_wrapper.cc b/lib/tcpip/network_wrapper.cc index 97553298..9bb3fae6 100644 --- a/lib/tcpip/network_wrapper.cc +++ b/lib/tcpip/network_wrapper.cc @@ -422,7 +422,7 @@ int SealedSocket::signal_event_futex(SocketEventType type) { return -EINVAL; } - uint32_t current = futex.load(); + int32_t current = futex.load(); while (current != SocketNotAvailable) { /* @@ -443,9 +443,9 @@ int SealedSocket::signal_event_futex(SocketEventType type) int SealedSocket::consume_event_futex(SocketEventType type) { - auto &futex = eventFutexState[type]; - uint32_t current = futex.load(); - while (current != SocketNotAvailable && current != 0) + auto &futex = eventFutexState[type]; + int32_t current = futex.load(); + while (current != SocketNotAvailable) { if (futex.compare_exchange_strong(current, current - 1)) { @@ -453,7 +453,7 @@ int SealedSocket::consume_event_futex(SocketEventType type) } } - return (current == SocketNotAvailable) ? -EINVAL : 0; + return -EINVAL; } /** @@ -933,18 +933,17 @@ uint32_t *network_socket_get_event_source(Socket sealedSocket, { with_sealed_socket( [&](SealedSocket *socket) { - auto *futex = - reinterpret_cast(&socket->eventFutexState[type]); - - Capability readyOnlyEventSource{futex}; - // Restrict the capability to read-only so the futex can only be - // modified through the callback, not by the caller of this - // helper. - readyOnlyEventSource.bounds() = sizeof(uint32_t); - readyOnlyEventSource.permissions() &= + auto *futex = &socket->eventFutexState[type]; + + Capability readOnlyEventSource{futex}; + // Expose the futex as read-only. The caller may observe its raw + // 32-bit representation but may modify it only through the + // socket callbacks. + readOnlyEventSource.bounds() = sizeof(*futex); + readOnlyEventSource.permissions() &= {Permission::Load, Permission::Global}; - result = readyOnlyEventSource.get(); + result = reinterpret_cast(readOnlyEventSource.get()); return 0; }, sealedSocket); diff --git a/lib/tcpip/tcpip-internal.h b/lib/tcpip/tcpip-internal.h index 98cfd3c4..825bced4 100644 --- a/lib/tcpip/tcpip-internal.h +++ b/lib/tcpip/tcpip-internal.h @@ -49,7 +49,7 @@ struct SealedSocket * feature. Different events increment different futexes in the array and * wake the corresponding waiting threads. */ - std::atomic eventFutexState[NumFutexTypes]; + std::atomic eventFutexState[NumFutexTypes]; /** * Increments the futex and notifies all waiters if the futex is still * valid. @@ -58,16 +58,17 @@ struct SealedSocket */ int signal_event_futex(SocketEventType type); /** - * Consume one pending event by decrementing the futex. + * Records one successfully accepted child socket by decrementing the accept + * event counter. * * The caller must hold `socketLock`, which prevents the socket from being - * deallocated while the futex is accessed. This operation is only - * bookkeeping: the caller has already consumed the corresponding event, so - * a zero counter is not an error. + * freed while the futex is accessed. The futex value may become negative + * if this helper is invoked after `FreeRTOS_accept()` successfully dequeue + * a child, but before `on_tcp_connect()` increment the futex. The delayed + * increment will repay this temporary debt. * - * Returns 0 if the futex remains valid, regardless of whether the counter - * was decremented. Returns `-EINVAL` if the futex has been invalidated by - * being set to `SocketNotAvailable` because the socket is being torn down. + * Returns 0 on success, or `-EINVAL` if the futex has been invalidated by + * being set to `SocketNotAvailable`. */ int consume_event_futex(SocketEventType type); /** From 1406be47e4c80f546f671964d64613874c2536a2 Mon Sep 17 00:00:00 2001 From: nding04 Date: Sat, 8 Aug 2026 20:27:45 -0700 Subject: [PATCH 4/6] TCPIP: initialize accepted socket wrappers --- include/NetAPI.h | 8 ++++++-- lib/tcpip/network_wrapper.cc | 8 ++++++++ 2 files changed, 14 insertions(+), 2 deletions(-) diff --git a/include/NetAPI.h b/include/NetAPI.h index e6f4f00b..8f418619 100644 --- a/include/NetAPI.h +++ b/include/NetAPI.h @@ -40,8 +40,12 @@ struct NetworkAddress */ enum SocketEventType : uint8_t { - SocketAcceptEvent = 0, // Triggered when a new TCP connection is accepted on - // a listening socket + /// Triggered when a new TCP connection + /// is accepted on a listening socket. + /// The value of the futex corresponding to + /// SocketAcceptEvent, means the number of + /// pending connections. + SocketAcceptEvent = 0, // SocketReceiveEvent = 1, // Triggered when data is received on a socket // SocketSendEvent = 2 // Triggered when a socket has space available // for sending diff --git a/lib/tcpip/network_wrapper.cc b/lib/tcpip/network_wrapper.cc index 9bb3fae6..ae310fd7 100644 --- a/lib/tcpip/network_wrapper.cc +++ b/lib/tcpip/network_wrapper.cc @@ -715,6 +715,13 @@ Socket network_socket_accept_tcp(Timeout *timeout, } socketWrapper->socketEpoch = currentSocketEpoch.load(); + /* + * For the newly allocated child socket, initialize the futexes. + */ + for (auto &eventState : socketWrapper->eventFutexState) + { + eventState.store(0); + } struct freertos_sockaddr addressTmp; uint32_t addressLength = sizeof(addressTmp); @@ -798,6 +805,7 @@ Socket network_socket_accept_tcp(Timeout *timeout, token_obj_destroy(mallocCapability, socket_key(), sealedSocket); return -EINVAL; } + xSocketSetSocketID(rawSocket, socketWrapper); // Set `address`. if ((heap_claim_ephemeral(timeout, address) < 0) || From 4ae58cc4fc6c9835a6751f4f3c53045c7ece7777 Mon Sep 17 00:00:00 2001 From: nding04 Date: Sun, 9 Aug 2026 23:59:41 -0700 Subject: [PATCH 5/6] TCPIP: add multiwaiter support for TCP send Add a TCP send event for multiwaiter, so one thread can wait for free send space on several connected TCP sockets at once. Workflow: The caller first calls `network_socket_send()` with a zero timeout. The first non-empty send creates `txStream` and sets the send event to its exact free space. A send queues as many bytes as it can and subtracts that count from the send event futex. When the peer ACKs bytes, `on_tcp_sent()` adds that count to the futex and wakes all waiters. If more bytes remain, the caller waits with multiwaiter and then tries the send again. If the connection closes, the waiter wakes and its next call to any network API will returns `-ENOTCONN`. Key changes: - Add `SocketTCPSendEvent`, whose corresponding futex represents the free byte count in `txStream`. - Keep `txStream` setup lazy just like FreeRTOS_send(), and use `FreeRTOS_tx_space()` for its usable size. - Register the sent callback for TCP clients and listeners. Because accepted sockets inherit it from their listener, although the listening socket itself cannot send application bytes. - Add `SocketConnectionClosed` for a disconnected TCP state. Keep `SocketNotAvailable` for a socket wrapper that is being removed. - On disconnect, set the send event futex to `SocketConnectionClosed` and wake its waiters. --- include/NetAPI.h | 24 ++++++-- lib/tcpip/network_wrapper.cc | 106 ++++++++++++++++++++++++++++++++--- lib/tcpip/tcpip-internal.h | 23 +++++--- 3 files changed, 131 insertions(+), 22 deletions(-) diff --git a/include/NetAPI.h b/include/NetAPI.h index 8f418619..f010ab5f 100644 --- a/include/NetAPI.h +++ b/include/NetAPI.h @@ -46,13 +46,26 @@ enum SocketEventType : uint8_t /// SocketAcceptEvent, means the number of /// pending connections. SocketAcceptEvent = 0, - // SocketReceiveEvent = 1, // Triggered when data is received on a socket - // SocketSendEvent = 2 // Triggered when a socket has space available - // for sending + /// Triggered when a socket has space available + /// for sending bytes. + /// The value of the futex corresponding to + /// SocketTCPSendEvent, means the number of free + /// bytes available in the txStream buffer of + /// the socket. + /// After multiwaiter_wait() returns, do not wait for this value to reach + /// the full size of the pending send. Call network_socket_send() again + /// whenever any space is available; it may send only part of the requested + /// data. + SocketTCPSendEvent = 1, + // Triggered when data is received on a socket + // SocketReceiveEvent = 2, }; /// Number of distinct socket event futex types. -static constexpr size_t NumFutexTypes = SocketEventType::SocketAcceptEvent + 1; +static constexpr size_t NumFutexTypes = SocketEventType::SocketTCPSendEvent + 1; +/// Sentinel value stored in the TCP send event futex when the wrapper still +/// exists but the TCP connection can no longer send. +static constexpr int32_t SocketConnectionClosed = INT32_MIN + 1; /// Sentinel value stored in a socket event futex after the socket has been /// torn down. static constexpr int32_t SocketNotAvailable = INT32_MIN; @@ -296,6 +309,9 @@ Socket __cheri_compartment("TCPIP") * Return the event source associated with a socket. * * The returned capability is read-only and bounded to four bytes. + * For a connected TCP socket, the first non-empty send initializes + * `SocketTCPSendEvent`, which then counts bytes that a zero-timeout send may + * add to the transmit stream. */ uint32_t *__cheri_compartment("TCPIP") network_socket_get_event_source(Socket sealedSocket, SocketEventType type); diff --git a/lib/tcpip/network_wrapper.cc b/lib/tcpip/network_wrapper.cc index ae310fd7..48c0ede0 100644 --- a/lib/tcpip/network_wrapper.cc +++ b/lib/tcpip/network_wrapper.cc @@ -415,7 +415,7 @@ namespace } // namespace -int SealedSocket::signal_event_futex(SocketEventType type) +int SealedSocket::signal_event_futex(SocketEventType type, int32_t count) { auto &futex = eventFutexState[type]; if (heap_claim_ephemeral(TimeoutWaitForever, &futex) != 0) @@ -423,15 +423,16 @@ int SealedSocket::signal_event_futex(SocketEventType type) return -EINVAL; } int32_t current = futex.load(); - while (current != SocketNotAvailable) + while ((current != SocketConnectionClosed) && + (current != SocketNotAvailable)) { /* * If the futex's current value equals current, - * store current + 1 and return true. Otherwise + * store current + count and return true. Otherwise * write the actual current value into current * (by reference) and return false. */ - if (futex.compare_exchange_strong(current, current + 1)) + if (futex.compare_exchange_strong(current, current + count)) { futex.notify_all(); return 0; @@ -440,14 +441,41 @@ int SealedSocket::signal_event_futex(SocketEventType type) return -EINVAL; } +/** + * Wake send waiters when FreeRTOS reports that the TCP connection has closed. + * Preserve SocketNotAvailable if the socket wrapper is already being torn down. + */ +void SealedSocket::mark_tcp_send_closed() +{ + auto &futex = eventFutexState[SocketEventType::SocketTCPSendEvent]; + // This function will only be called from FreeRTOS callback, + // so the caller will not held the socket lock when calling this + // function. Thus, ephemeral call is needed to prevent Use-After-Free + // issue. + if (heap_claim_ephemeral(TimeoutWaitForever, &futex) != 0) + { + return; + } + int32_t current = futex.load(); + while ((current != SocketConnectionClosed) && + (current != SocketNotAvailable)) + { + if (futex.compare_exchange_strong(current, SocketConnectionClosed)) + { + futex.notify_all(); + return; + } + } +} -int SealedSocket::consume_event_futex(SocketEventType type) +int SealedSocket::consume_event_futex(SocketEventType type, int32_t count) { auto &futex = eventFutexState[type]; int32_t current = futex.load(); - while (current != SocketNotAvailable) + while ((current != SocketConnectionClosed) && + (current != SocketNotAvailable)) { - if (futex.compare_exchange_strong(current, current - 1)) + if (futex.compare_exchange_strong(current, current - count)) { return 0; } @@ -455,11 +483,30 @@ int SealedSocket::consume_event_futex(SocketEventType type) return -EINVAL; } +/** + * Callback called by FreeRTOS+TCP when TCP bytes are acknowledged. + * + * Add the acknowledged byte count to the send event futex and wake all threads + * waiting for txStream space. + */ +static void on_tcp_sent(Socket_t socket, size_t length) +{ + // Retrieve the wrapper from the back pointer. + auto *wrapper = static_cast(pvSocketGetSocketID(socket)); + if (wrapper != nullptr) + { + wrapper->signal_event_futex(SocketEventType::SocketTCPSendEvent, + static_cast(length)); + } +} +F_TCP_UDP_Handler_t onTCPSentCallback = {nullptr, nullptr, on_tcp_sent}; /** * Callback called by FreeRTOS+TCP when a TCP connection is created or * terminated. * + * A terminated connection also wakes threads waiting for TCP send space. + * * We use this callback to handle the case where a three-way TCP handshake * initiated by a peer on a listening socket fails. * @@ -509,8 +556,14 @@ static void on_tcp_connect(Socket_t socket, BaseType_t isConnected) firewall_remove_tcpipv4_remote_endpoint( address.sin_address.ulIP_IPv4, localPort, address.sin_port); } + auto *wrapper = + static_cast(pvSocketGetSocketID(socket)); + if (wrapper != nullptr) + { + wrapper->mark_tcp_send_closed(); + } } - else + else if (socket->u.xTCP.eTCPState == eTCP_LISTEN) { // Update eventFutexState in the listening socket and wake up the // threads sleeping on the corresponding futex. @@ -664,13 +717,25 @@ Socket network_socket_create_and_bind(Timeout *timeout, mallocCapability, socket_key(), sealedSocket); return -EAGAIN; } - + } + if (type == ConnectionTypeTCP) + { FreeRTOS_setsockopt( socket, 0, FREERTOS_SO_TCP_CONN_HANDLER, static_cast(&onTCPConnectCallback), sizeof(onTCPConnectCallback)); + /* + * Register the callback `on_tcp_sent()` + * for sending bytes. This is to support + * multiwaiter feature in network stack. + */ + FreeRTOS_setsockopt(socket, + 0, + FREERTOS_SO_TCP_SENT_HANDLER, + static_cast(&onTCPSentCallback), + sizeof(onTCPSentCallback)); } } else @@ -1421,14 +1486,37 @@ ssize_t network_socket_send(Timeout *timeout, { return -EPERM; } + // Create the txStream and initialize send futex on the + // first non-zero send. + if ((length != 0) && (socket->socket->u.xTCP.txStream == nullptr)) + { + if (FreeRTOS_get_tx_base(socket->socket) == nullptr) + { + // Allocation failed. + return -ENOMEM; + } + // Store current available bytes to initialize the futex. + auto &sendFutex = + socket->eventFutexState[SocketEventType::SocketTCPSendEvent]; + int32_t uninitialized = 0; + // The reason why we use compare_exchange_* here is that + // if the socket is disconnected, a plain .store() would + // overwrite SocketConnectionClosed. + sendFutex.compare_exchange_strong( + uninitialized, FreeRTOS_tx_space(socket->socket)); + } Debug::log("Sending {}-byte TCP packet from {}", length, buffer); int ret = with_freertos_timeout( timeout, socket->socket, FREERTOS_SO_SNDTIMEO, [&] { return FreeRTOS_send(socket->socket, buffer, length, 0); }); Debug::log("FreeRTOS_send returned {}", ret); + // Subtract the bytes queued by FreeRTOS_send() from the available + // txStream space. if (ret >= 0) { + socket->consume_event_futex(SocketEventType::SocketTCPSendEvent, + ret); return ret; } if (ret == -pdFREERTOS_ERRNO_ENOTCONN) diff --git a/lib/tcpip/tcpip-internal.h b/lib/tcpip/tcpip-internal.h index 825bced4..a30a5c75 100644 --- a/lib/tcpip/tcpip-internal.h +++ b/lib/tcpip/tcpip-internal.h @@ -51,15 +51,20 @@ struct SealedSocket */ std::atomic eventFutexState[NumFutexTypes]; /** - * Increments the futex and notifies all waiters if the futex is still - * valid. + * Increments the futex by `count` and notifies all waiters if the futex is + * still valid. * - * Returns 0 on success, or `-EINVAL` if the futex has been invalidated. + * Returns 0 on success, or `-EINVAL` if the futex has been set to a + * terminal value. */ - int signal_event_futex(SocketEventType type); + int signal_event_futex(SocketEventType type, int32_t count = 1); /** - * Records one successfully accepted child socket by decrementing the accept - * event counter. + * Marks the TCP send event futex as closed and notifies all waiters. + */ + void mark_tcp_send_closed(); + /** + * Records successfully consumed events by decrementing the event counter by + * `count`. * * The caller must hold `socketLock`, which prevents the socket from being * freed while the futex is accessed. The futex value may become negative @@ -67,10 +72,10 @@ struct SealedSocket * a child, but before `on_tcp_connect()` increment the futex. The delayed * increment will repay this temporary debt. * - * Returns 0 on success, or `-EINVAL` if the futex has been invalidated by - * being set to `SocketNotAvailable`. + * Returns 0 on success, or `-EINVAL` if the futex has been set to a + * terminal value. */ - int consume_event_futex(SocketEventType type); + int consume_event_futex(SocketEventType type, int32_t count = 1); /** * The lock protecting this socket. */ From ec84b93254300f385129181ae6cae089ed1ec981 Mon Sep 17 00:00:00 2001 From: nding04 Date: Sat, 15 Aug 2026 17:39:50 -0700 Subject: [PATCH 6/6] Add futex initialization helper. Added the futex initialization helper to avoid code duplication. The send futex is initialized to the maximum txStream size before txStream is initialized. This is to prevent the user from calling multiwaiter_wait before calling the first send, which would result in waiting forever if we initialized it to 0. Later, on the first send, FreeRTOS initializes the txStream buffer, and the futex value is updated to the exact value instead of an advertised placeholder value. --- include/NetAPI.h | 7 +++--- lib/tcpip/network_wrapper.cc | 45 +++++++++++++++++++++--------------- lib/tcpip/tcpip-internal.h | 9 ++++++++ 3 files changed, 39 insertions(+), 22 deletions(-) diff --git a/include/NetAPI.h b/include/NetAPI.h index f010ab5f..c5dd8b2d 100644 --- a/include/NetAPI.h +++ b/include/NetAPI.h @@ -309,9 +309,10 @@ Socket __cheri_compartment("TCPIP") * Return the event source associated with a socket. * * The returned capability is read-only and bounded to four bytes. - * For a connected TCP socket, the first non-empty send initializes - * `SocketTCPSendEvent`, which then counts bytes that a zero-timeout send may - * add to the transmit stream. + * For a TCP socket, the futex `SocketTCPSendEvent` initially reports + * the configured maximum txStream capacity before the buffer is allocated. + * This is to prevent the user thread from sleeping on the multi-waiter + * forever before the first `network_socket_send()` is called. */ uint32_t *__cheri_compartment("TCPIP") network_socket_get_event_source(Socket sealedSocket, SocketEventType type); diff --git a/lib/tcpip/network_wrapper.cc b/lib/tcpip/network_wrapper.cc index 48c0ede0..0ae553ee 100644 --- a/lib/tcpip/network_wrapper.cc +++ b/lib/tcpip/network_wrapper.cc @@ -415,6 +415,24 @@ namespace } // namespace +void SealedSocket::initialize_event_futexes() +{ + for (auto &eventState : eventFutexState) + { + eventState.store(0); + } + + if (socket->ucProtocol == FREERTOS_IPPROTO_TCP) + { + /* + * FreeRTOS creates txStream lazily. Lie about its current space and + * use its maximum size now: otherwise a caller that waits on this futex + * before its first send() will block forever. + */ + eventFutexState[SocketTCPSendEvent].store(FreeRTOS_tx_space(socket)); + } +} + int SealedSocket::signal_event_futex(SocketEventType type, int32_t count) { auto &futex = eventFutexState[type]; @@ -608,12 +626,6 @@ Socket network_socket_create_and_bind(Timeout *timeout, // Set the socket epoch socketWrapper->socketEpoch = currentSocketEpoch.load(); - // Multi-waiter: initialize the futexes. - for (auto &eventState : socketWrapper->eventFutexState) - { - eventState.store(0); - } - const auto Family = isIPv6 ? FREERTOS_AF_INET6 : FREERTOS_AF_INET; Socket_t socket = FreeRTOS_socket(Family, @@ -637,6 +649,7 @@ Socket network_socket_create_and_bind(Timeout *timeout, * socket wrapper from the FreeRTOS socket in the callbacks. */ socketWrapper->socket = socket; + socketWrapper->initialize_event_futexes(); xSocketSetSocketID(socket, socketWrapper); // Claim the socket so that it counts towards the caller's quota. The @@ -780,14 +793,6 @@ Socket network_socket_accept_tcp(Timeout *timeout, } socketWrapper->socketEpoch = currentSocketEpoch.load(); - /* - * For the newly allocated child socket, initialize the futexes. - */ - for (auto &eventState : socketWrapper->eventFutexState) - { - eventState.store(0); - } - struct freertos_sockaddr addressTmp; uint32_t addressLength = sizeof(addressTmp); FreeRTOS_Socket_t *rawSocket = nullptr; @@ -836,6 +841,7 @@ Socket network_socket_accept_tcp(Timeout *timeout, return futexConsumeResult; } socketWrapper->socket = rawSocket; + socketWrapper->initialize_event_futexes(); // Claim the socket so that it counts towards the caller's quota. The // network stack also keeps a claim to it. We will drop this claim on @@ -1486,24 +1492,25 @@ ssize_t network_socket_send(Timeout *timeout, { return -EPERM; } - // Create the txStream and initialize send futex on the - // first non-zero send. + // Create txStream on the first non-zero send and replace the + // fake place holder value with its exact available space. if ((length != 0) && (socket->socket->u.xTCP.txStream == nullptr)) { + int32_t advertisedSpace = FreeRTOS_tx_space(socket->socket); if (FreeRTOS_get_tx_base(socket->socket) == nullptr) { // Allocation failed. return -ENOMEM; } - // Store current available bytes to initialize the futex. + // Reconcile the advertised capacity with the newly allocated + // stream's exact usable space. auto &sendFutex = socket->eventFutexState[SocketEventType::SocketTCPSendEvent]; - int32_t uninitialized = 0; // The reason why we use compare_exchange_* here is that // if the socket is disconnected, a plain .store() would // overwrite SocketConnectionClosed. sendFutex.compare_exchange_strong( - uninitialized, FreeRTOS_tx_space(socket->socket)); + advertisedSpace, FreeRTOS_tx_space(socket->socket)); } Debug::log("Sending {}-byte TCP packet from {}", length, buffer); int ret = with_freertos_timeout( diff --git a/lib/tcpip/tcpip-internal.h b/lib/tcpip/tcpip-internal.h index a30a5c75..35485b8e 100644 --- a/lib/tcpip/tcpip-internal.h +++ b/lib/tcpip/tcpip-internal.h @@ -50,6 +50,15 @@ struct SealedSocket * wake the corresponding waiting threads. */ std::atomic eventFutexState[NumFutexTypes]; + /** + * Initialize all event futexes for a given socket. + * This should be called as long as a socket is created. + * All futexes will be initialized to `0` to align with + * the futexes semantics, except `SocketTCPSendEvent` futex. + * It will be initialized to pre-configured txStream buffer + * size to prevent indefinite blocking thread. + */ + void initialize_event_futexes(); /** * Increments the futex by `count` and notifies all waiters if the futex is * still valid.