Skip to content
Merged
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
1 change: 1 addition & 0 deletions _local_deps
2 changes: 1 addition & 1 deletion doc/benchmark.md
Original file line number Diff line number Diff line change
Expand Up @@ -223,7 +223,7 @@ RSS(MiB) = 0.0008870 × N + 10.120 R² = 0.9997352
散布」,而不是两次独立读数的绝对值。

**试过但没采纳**:把读缓冲封顶到 8 KiB(−0.49%,落在散布内,白搭)。而且**不能封在
`alloc_buf()` 里** —— 它同时给 UDP 用(`src/net/uvcpp_udp_server.cpp:165`、
`alloc_buf()` 里** —— 它同时给 UDP 用(`src/net/uvcpp_udp_server.cpp:185-185`、
`src/net/uvcpp_udp_client.cpp:1148-1148`),而 UDP 是数据报,缓冲小了会**截断报文**。

### 4.2 接受连接时默认关 Nagle(**另一套装置**,轮流 A/B)
Expand Down
2 changes: 1 addition & 1 deletion doc/net-guide.md
Original file line number Diff line number Diff line change
Expand Up @@ -719,7 +719,7 @@ int main() {
```

- `recv_start` 的回调里那个 `uvcpp_buf*` 是**栈上的临时量**,负载是克隆出来的,
底层读缓冲回调一返回就释放(`src/net/uvcpp_udp_server.cpp:177-201`)。**不要留那个指针。**
底层读缓冲回调一返回就释放(`src/net/uvcpp_udp_server.cpp:197-221`)。**不要留那个指针。**
- `send(ip, port, data, len, cb = nullptr)` 有**同一个 nullptr 陷阱**:省略 `cb` 就是
同步发送。
- 客户端侧另有 `recv_wait(out_buf, timeout_ms)` 和 `set_max_read_cache_size()`,
Expand Down
40 changes: 31 additions & 9 deletions src/net/uvcpp_udp_server.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
#include <uvcpp/uvcpp_define.h>

#include <chrono>
#include <memory>
#include <cstring>
#include <stdexcept>
#include <thread>
Expand Down Expand Up @@ -38,6 +39,13 @@ uvcpp_udp_server::uvcpp_udp_server() {
status_ = UDP_SERVER_NONE;
}

uvcpp_udp_server::uvcpp_udp_server(uvcpp_loop* loop) {
loop_ = loop != nullptr ? loop : uvcpp_loop::default_loop();
udp_ = new uvcpp_udp(loop_);
owns_loop_ = false;
status_ = UDP_SERVER_NONE;
}

std::shared_ptr<char> uvcpp_udp_server::alive_token() {
if (!alive_token_) alive_token_.reset(new char(0));
return alive_token_;
Expand Down Expand Up @@ -72,13 +80,22 @@ uvcpp_udp_server::~uvcpp_udp_server() {
// —— `recv_stop()` 会把活跃计数减掉,带上它就会把"读过、还开着、已经不
// 在读"误判成"早就关完了",于是 socket 不关、句柄留在队列上。
if (!udp_->is_closing()) {
bool close_done = false;
udp_->close([&close_done](uvcpp_handle*) { close_done = true; });
for (int i = 0; i < 64 && !close_done; ++i) {
if (loop_ != nullptr) loop_->run(UV_RUN_NOWAIT);
if (owns_loop_) {
// `close()` 的回调是**可能晚于本析构**才跑的(64 轮泵不完就留到调用方
// 那一拍),所以完成标志不能是栈上的引用 —— 那把回调会写进一个已经退栈
// 的帧。共享指针让迟到的回调有一个活着的落点;我们不等它。
std::shared_ptr<bool> close_done = std::make_shared<bool>(false);
udp_->close([close_done](uvcpp_handle*) { *close_done = true; });
for (int i = 0; i < 64 && !*close_done; ++i) {
if (loop_ != nullptr) loop_->run(UV_RUN_NOWAIT);
}
}
} else {
if (loop_ != nullptr) loop_->run(UV_RUN_NOWAIT);
// 外部循环:**不显式 close、不泵**(泵别人的循环会替调用方跑循环线程,
// 而上面那种带栈上落点的回调在这里必然迟到)。`delete udp_` 的
// `free_handle()` 会换成 detached marker 再走 `uv_close` —— 底层是
// `uv_close` 之后的收尾,不需要回调里的任何东西,由循环属主那一拍收场。
} else if (owns_loop_ && loop_ != nullptr) {
loop_->run(UV_RUN_NOWAIT);
}
delete udp_;
udp_ = nullptr;
Expand All @@ -91,14 +108,17 @@ uvcpp_udp_server::~uvcpp_udp_server() {
// 后者内部会 `stop()`,而 `uv_run` 的
// `while (r != 0 && loop->stop_flag == 0)` 是**进 body 之前**判的,停标志一
// 立那一轮就整段空转,交替调用等于原地打转。
if (loop_ != nullptr) {
// 外部循环整段跳过(`owns_loop_ == false`):循环不是我们的,关它等于把同一条
// 循环上别的对象的句柄连同它们的事件一起废掉,`delete loop_` 更是直接还别人的
// 内存。句柄该关的上面已经关过了,收尾留给循环的属主。
if (loop_ != nullptr && owns_loop_) {
for (int i = 0; i < 256 && loop_->loop_alive() != 0; ++i) {
loop_->run(UV_RUN_NOWAIT);
}
loop_->loop_close();
delete loop_;
loop_ = nullptr;
}
loop_ = nullptr;

if (recv_arg_ != nullptr) {
delete static_cast<std::function<void(uvcpp_buf*, const char*, int)>*>(
Expand Down Expand Up @@ -379,7 +399,9 @@ void uvcpp_udp_server::stop(std::function<void()> on_stopped) {
// =========================================================================

int uvcpp_udp_server::run(uv_run_mode md) { return loop_->run(md); }
void uvcpp_udp_server::stop_loop() { loop_->stop(); }
void uvcpp_udp_server::stop_loop() {
if (owns_loop_ && loop_ != nullptr) loop_->stop();
}

// =========================================================================
// Internal
Expand Down
15 changes: 15 additions & 0 deletions src/net/uvcpp_udp_server.h
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,18 @@ class UVCPP_API uvcpp_udp_server {
UVCPP_DEFINE_FUNC(uvcpp_udp_server)
UVCPP_DEFINE_COPY_FUNC_DELETE(uvcpp_udp_server)

/**
* @brief 用一条**外部**事件循环,与 `uvcpp_tcp_server(uvcpp_loop*)` 同一约定:
* 本对象**不拥有**它,析构时也不关不删;`run()` 由调用方负责。
*
* issue #59:`uvcpp_web_app` / 中继这类"一条循环上既有
* WS/TCP 又要 UDP 端点"的场景,自带循环的版本会把回调
* 放到另一条线程上,跨到循环亲和的连接上就不安全。
*
* @warning 外部循环必须活得比本对象长。`stop_loop()` 在外部模式下是**空操作**。
*/
explicit uvcpp_udp_server(uvcpp_loop* loop);

// -----------------------------------------------------------------
// Accessors
// -----------------------------------------------------------------
Expand Down Expand Up @@ -109,6 +121,7 @@ class UVCPP_API uvcpp_udp_server {
// Loop control
// -----------------------------------------------------------------

/** @brief 外部循环模式下不做任何事(那条循环不是本对象的)。 */
int run(uv_run_mode md = UV_RUN_DEFAULT);
void stop_loop();

Expand Down Expand Up @@ -138,6 +151,8 @@ class UVCPP_API uvcpp_udp_server {

uvcpp_loop* loop_ = nullptr;
uvcpp_udp* udp_ = nullptr;
/// 外部循环构造进来时置 false:析构不关、不删它。
bool owns_loop_ = true;
int status_ = UDP_SERVER_NONE;
int last_error_code_ = 0;

Expand Down
175 changes: 175 additions & 0 deletions tests/functional/udp_server_external_loop_func.cpp
Original file line number Diff line number Diff line change
@@ -0,0 +1,175 @@
/**
* @file tests/functional/udp_server_external_loop_func.cpp
* @brief issue #59:`uvcpp_udp_server` 用**外部循环**构造时的语义。
*
* 背景:`uvcpp_tcp_server` / `uvcpp_quic_server` 都支持"外部循环 + 在既有循环上
* 工作",`uvcpp_udp_server` 原先只能自带一条循环 —— 想在 `uvcpp_web_app` 所在的
* 循环上再收一个 UDP 端点(中继数据面)就做不到,回调会被放到另一条线程上,
* 跨到循环亲和的 WS/TCP/QUIC 连接上就不安全。
*
* 判据:
* [1] `uvcpp_udp_server(uvcpp_loop*)` 认下这条循环(`get_loop()` 返回同一个
* 指针),收发都在它上面;**循环由调用方 `run()`**。
* [2] `stop()` 只关本端点的句柄,**不停**外部循环:同一条循环上还能再起第二个
* 端点,并正常收发。
* [3] 析构不关、不删外部循环 —— 这里用的是**栈上的** `uvcpp_loop`:真去
* `delete`/`loop_close` 它会当场崩(glibc 的 "free(): invalid pointer" 或
* ASAN 的 stack-use-after-scope),不需要额外断言。
*/
#ifdef _WIN32
#include <winsock2.h>
#include <ws2tcpip.h>
#else
#include <arpa/inet.h>
#include <netinet/in.h>
#include <sys/socket.h>
#include <unistd.h>
#endif

#include <atomic>
#include <cstdio>
#include <cstring>
#include <functional>
#include <string>

#include <handle/uvcpp_loop.h>
#include <handle/uvcpp_timer.h>
#include <net/uvcpp_udp_server.h>
#include <uvcpp/uvcpp_buf.h>

using namespace uvcpp;

namespace {

int g_failures = 0;

void check(bool ok, const char* what) {
std::printf(" %s %s\n", ok ? "[ok]" : "[FAIL]", what);
if (!ok) ++g_failures;
}

/** 把一条已 bind 的 UDP 端点上的端口取出来。 */
int bound_port(uvcpp_udp_server& srv) {
struct sockaddr_in name;
int namelen = sizeof(name);
if (srv.get_udp()->getsockname(reinterpret_cast<sockaddr*>(&name),
&namelen) != 0) {
return -1;
}
return ntohs(name.sin_port);
}

/** 裸 UDP socket 发一段;不经过 libuv,所以与循环无关。 */
bool send_raw(int port, const char* text) {
#ifdef _WIN32
// 进程级 Winsock 初始化一次即可;这个测试只发一两个包,不进 I/O 完成端口。
static const bool wsa_ok = []() {
WSADATA d;
return WSAStartup(MAKEWORD(2, 2), &d) == 0;
}();
if (!wsa_ok) return false;
SOCKET fd = ::socket(AF_INET, SOCK_DGRAM, 0);
if (fd == INVALID_SOCKET) return false;
#else
int fd = ::socket(AF_INET, SOCK_DGRAM, 0);
if (fd < 0) return false;
#endif
struct sockaddr_in addr;
std::memset(&addr, 0, sizeof(addr));
addr.sin_family = AF_INET;
addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
addr.sin_port = htons(static_cast<uint16_t>(port));
const int len = static_cast<int>(std::strlen(text));
const int sent =
static_cast<int>(::sendto(fd, text, len, 0,
reinterpret_cast<sockaddr*>(&addr),
static_cast<int>(sizeof(addr))));
const bool ok = sent == len;
#ifdef _WIN32
::closesocket(fd);
#else
::close(fd);
#endif
return ok;
}

/** 有界地泵调用方的循环,直到 done() 为真。 */
bool pump_until(uvcpp_loop& loop, const std::function<bool()>& done,
int rounds = 4000) {
for (int i = 0; i < rounds; ++i) {
if (done()) return true;
loop.run(UV_RUN_NOWAIT);
}
return done();
}

void test_external_loop_receives_and_survives_stop() {
std::printf("[udp_server_external] external_loop\n");

uvcpp_loop loop; // 栈上:若被 delete 会当场崩
std::string seen;
uvcpp_udp_server server(&loop);

check(server.get_loop() == &loop, "get_loop() 返回传进去的那条循环");
check(server.bind("127.0.0.1", 0) == 0, "bind 成功");
const int port = bound_port(server);
check(port > 0, "拿到了绑定端口");

server.recv_start([&seen](uvcpp_buf* buf, const char*, int) {
seen.assign(buf->get_data(), buf->size());
});

check(send_raw(port, "hello-external"), "裸 socket 发出第一包");
check(pump_until(loop, [&seen] { return seen == "hello-external"; }),
"回调在调用方的循环上被派发(调用方 run)");

// stop() 只关本端点;外部循环不许被停掉。
std::atomic<bool> stopped{false};
server.stop([&stopped] { stopped.store(true); });
check(pump_until(loop, [&stopped] { return stopped.load(); }),
"stop() 的关闭回合同样落在外部循环上");

// 同一条循环上再起一个端点,仍然能收 —— 证明循环没被 stop()/析构带走。
std::string seen2;
uvcpp_udp_server server2(&loop);
check(server2.bind("127.0.0.1", 0) == 0, "同一循环上第二个端点 bind 成功");
const int port2 = bound_port(server2);
server2.recv_start([&seen2](uvcpp_buf* buf, const char*, int) {
seen2.assign(buf->get_data(), buf->size());
});
check(send_raw(port2, "second-endpoint"), "发到第二个端点");
check(pump_until(loop, [&seen2] { return seen2 == "second-endpoint"; }),
"第二个端点在同一循环上收到(stop() 没有停外部循环)");

// 外部模式下 stop_loop() 是空操作:调完之后**第一轮** NOWAIT 就要派发。
//
// 判据必须是一轮:`uv_run` 在 `stop_flag` 立着时跳过整段轮询,返回前才把
// 标志清掉 —— "被停过"只吃掉一轮;给多轮的话停与不停都是绿的,判据空转。
//
// 用 0ms 定时器而不是 UDP 包:定时器在 `run` 之前就已到期,第一轮必然派发。
// 环回 UDP 的包在 macOS 的 kqueue 上不保证第一轮就可见(本用例在 macOS 的
// db 格子上就是这么红的)。定时器只证明"循环还在跑回调",与这一条判据正好
// 同一个命题,不牵扯收发时序。
bool timer_fired = false;
uvcpp_timer probe(&loop);
probe.start([&timer_fired](uvcpp_timer*) { timer_fired = true; }, 0, 0);
server2.stop_loop();
loop.run(UV_RUN_NOWAIT);
check(timer_fired,
"stop_loop() 在外部模式下是空操作(第一轮 NOWAIT 就派发)");
probe.stop();
}

} // namespace

int main() {
std::printf("[udp_server_external] issue #59\n");
test_external_loop_receives_and_survives_stop();

if (g_failures == 0) {
std::printf("[udp_server_external] ALL PASS\n");
return 0;
}
std::printf("[udp_server_external] FAIL (%d checks failed)\n", g_failures);
return 2;
}
4 changes: 2 additions & 2 deletions tests/tools/doc_line_refs.lock
Original file line number Diff line number Diff line change
Expand Up @@ -371,8 +371,8 @@ src/net/uvcpp_tcp_server.h:92-97 451a810cbb9be70b 30f2378010013043 a251d8a336202
src/net/uvcpp_tcp_server.h:965-965 0e77dae47f5f5e18 0e77dae47f5f5e18 -
src/net/uvcpp_udp_client.cpp:1148-1148 a0f156405122d91e a0f156405122d91e -
src/net/uvcpp_udp_client.h:74-74 25bf4d22cd8306d1 25bf4d22cd8306d1 -
src/net/uvcpp_udp_server.cpp:165-165 f756f35ae790d904 f756f35ae790d904 -
src/net/uvcpp_udp_server.cpp:177-201 b7593379f0bb1cb9 8e1d6d9580d5f162 2a9adda6eab309d9
src/net/uvcpp_udp_server.cpp:185-185 f756f35ae790d904 f756f35ae790d904 -
src/net/uvcpp_udp_server.cpp:197-221 b7593379f0bb1cb9 8e1d6d9580d5f162 2a9adda6eab309d9
src/net/uvcpp_udp_server.h:51-51 e1e48a6f8b9b57b8 e1e48a6f8b9b57b8 -
src/quic/uvcpp_quic_client.h:57-57 8a5a85c6bb9335c1 8a5a85c6bb9335c1 -
src/quic/uvcpp_quic_common.h:112-140 8b4637fd3ab9d5e5 f23a25e3225de0d2 476f0841a1ab03de
Expand Down
Loading