Skip to content

Commit 779a114

Browse files
chenxuan520cursoragent
chenxuan520
and
cursoragent
committed
refactor: replace thread with GCD dispatch_source for macOS timer
Co-authored-by: Cursor <cursoragent@cursor.com>
1 parent b73d16c commit 779a114

7 files changed

Lines changed: 150 additions & 42 deletions

File tree

‎src/cppnet/timer/timer.cpp‎

Lines changed: 53 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -40,35 +40,44 @@ int TimerSocket::Reset(int sec, int nsec) {
4040
return kSuccess;
4141

4242
#elif __APPLE__
43-
if (send_fd_ > 0) {
44-
Close();
43+
uint64_t interval_ns = (uint64_t)sec * NSEC_PER_SEC + (uint64_t)nsec;
44+
45+
// fast path: timer already running, just update interval in-place
46+
if (timer_ != nullptr) {
47+
dispatch_source_set_timer(timer_, DISPATCH_TIME_NOW, interval_ns, 0);
48+
return kSuccess;
4549
}
4650

51+
// first time: create pipe (read end exposed as Socket fd) + GCD timer
4752
int pipe_fd[2] = {-1, -1};
4853
if (pipe(pipe_fd) < 0) {
4954
return kSysErr;
5055
}
51-
5256
fd_ = pipe_fd[0];
5357
send_fd_ = pipe_fd[1];
54-
status_ = kInit;
5558

56-
auto timer_func = [=](int send_fd, int sec, int nsec) {
57-
uint64_t exp;
58-
TimerSocket soc(send_fd);
59-
while (true) {
60-
std::this_thread::sleep_for(
61-
std::chrono::nanoseconds((long long)(sec * 1e9) + nsec));
62-
auto rc = soc.Write(&exp, sizeof(uint64_t));
63-
if (rc <= 0) {
64-
// close socket
65-
break;
66-
}
67-
}
68-
};
69-
70-
thread_ = std::move(std::thread(timer_func, send_fd_, sec, nsec));
59+
timer_ = dispatch_source_create(
60+
DISPATCH_SOURCE_TYPE_TIMER, 0, 0,
61+
dispatch_get_global_queue(QOS_CLASS_USER_INITIATED, 0));
62+
if (timer_ == nullptr) {
63+
::close(fd_);
64+
::close(send_fd_);
65+
fd_ = -1;
66+
send_fd_ = -1;
67+
return kSysErr;
68+
}
69+
70+
// capture by value: send_fd_ may be reset to -1 in Close()
71+
int send_fd = send_fd_;
72+
dispatch_source_set_event_handler(timer_, ^{
73+
uint64_t exp = 1;
74+
::write(send_fd, &exp, sizeof(uint64_t));
75+
});
7176

77+
// leeway = 0 disables timer coalescing for higher precision
78+
dispatch_source_set_timer(timer_, DISPATCH_TIME_NOW, interval_ns, 0);
79+
dispatch_resume(timer_);
80+
status_ = kInit;
7281
return kSuccess;
7382
#else
7483
return kNotSupport;
@@ -77,11 +86,31 @@ int TimerSocket::Reset(int sec, int nsec) {
7786

7887
#ifdef __APPLE__
7988
int TimerSocket::Close() {
80-
Socket::Close();
81-
Socket(send_fd_).Close();
82-
thread_.join();
83-
send_fd_ = -1;
84-
return 0;
89+
if (timer_ != nullptr) {
90+
// dispatch_source_cancel is async; use semaphore to wait for the cancel
91+
// handler so the event handler can no longer write to send_fd_ after
92+
// Close() returns
93+
dispatch_semaphore_t sem = dispatch_semaphore_create(0);
94+
dispatch_source_set_cancel_handler(timer_, ^{
95+
dispatch_semaphore_signal(sem);
96+
});
97+
dispatch_source_cancel(timer_);
98+
dispatch_semaphore_wait(sem, DISPATCH_TIME_FOREVER);
99+
#if !OS_OBJECT_USE_OBJC
100+
dispatch_release(timer_);
101+
dispatch_release(sem);
102+
#endif
103+
timer_ = nullptr;
104+
}
105+
106+
if (send_fd_ >= 0) {
107+
::close(send_fd_);
108+
send_fd_ = -1;
109+
}
110+
if (status_ == kInit) {
111+
Socket::Close();
112+
}
113+
return kSuccess;
85114
}
86115
#endif
87116

‎src/cppnet/timer/timer.hpp‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@
22

33
#include "../socket/socket.hpp"
44
#ifdef __APPLE__
5-
#include <thread>
5+
#include <dispatch/dispatch.h>
66
#endif
77

88
namespace cppnet {
@@ -46,8 +46,10 @@ class TimerSocket : public Socket {
4646
int Close() override;
4747

4848
private:
49+
// pipe write end, driven by GCD timer; read end is fd_ in base Socket
4950
int send_fd_ = -1;
50-
std::thread thread_;
51+
// GCD timer source; nullptr before Init / after Close
52+
dispatch_source_t timer_ = nullptr;
5153
#endif
5254
};
5355

‎src/test/http/server/http_server_test.hpp‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,7 @@ using namespace std;
1111

1212
TEST(HttpServer, HelloWorld) {
1313
HttpServer server;
14-
Address addr{"127.0.0.1", 8080};
14+
Address addr{"127.0.0.1", 19981};
1515
auto rc = server.Init(addr);
1616
MUST_TRUE(rc == 0, server.err_msg());
1717

@@ -54,7 +54,7 @@ TEST(HttpServer, Static) {
5454
#ifdef _WIN32
5555
Address addr{"127.0.0.1", (uint16_t)(rand() % 1000 + 8000)};
5656
#else
57-
Address addr{"127.0.0.1", 8080};
57+
Address addr{"127.0.0.1", 19981};
5858
#endif
5959
auto rc = server.Init(addr);
6060
MUST_TRUE(rc == 0, server.err_msg());
@@ -173,7 +173,7 @@ TEST(HttpServer, Group) {
173173
#ifdef _WIN32
174174
Address addr{"127.0.0.1", (uint16_t)(rand() % 1000 + 8000)};
175175
#else
176-
Address addr{"127.0.0.1", 8080};
176+
Address addr{"127.0.0.1", 19981};
177177
#endif
178178
HttpServer server;
179179
auto rc = server.Init(addr);
@@ -219,7 +219,7 @@ TEST(HttpServer, Middleware) {
219219
#ifdef _WIN32
220220
Address addr{"127.0.0.1", (uint16_t)(rand() % 1000 + 8000)};
221221
#else
222-
Address addr{"127.0.0.1", 8080};
222+
Address addr{"127.0.0.1", 19981};
223223
#endif
224224
HttpServer server;
225225
auto rc = server.Init(addr);
@@ -309,7 +309,7 @@ TEST(HttpServer, HttpsServer) {
309309
return;
310310
}
311311
server.set_logger(std::make_shared<StdLogger>());
312-
Address addr{"127.0.0.1", 8080};
312+
Address addr{"127.0.0.1", 19981};
313313
std::shared_ptr<SSLContext> ssl_context = std::make_shared<SSLContext>();
314314
rc = ssl_context->InitSvrFile("./test/ssl/cert_demo.pem",
315315
"./test/ssl/cert_key.pem");

‎src/test/server/tcp_server_test.hpp‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,7 @@
77
using namespace cppnet;
88
using namespace std;
99

10-
Address addr{"127.0.0.1", 8080};
10+
Address addr{"127.0.0.1", 19981};
1111
TcpServer server{addr};
1212

1313
TEST(TcpServer, SigleClient) {
@@ -113,7 +113,7 @@ TEST(TcpServer, MultiClient) {
113113
string ip;
114114
uint16_t port = 0;
115115
addr_cli.GetIPAndPort(ip, port);
116-
MUST_TRUE(port != 8080, "port need not equal 8080");
116+
MUST_TRUE(port != 19981, "port need not equal 19981");
117117
DEBUG(fd.fd() << " accept " << ip << ":" << port);
118118
// accept
119119
} else if (event == TcpServer::kEventRead) {
@@ -217,7 +217,7 @@ TEST(TcpServer, MultiThread) {
217217
string ip;
218218
uint16_t port = 0;
219219
addr_cli.GetIPAndPort(ip, port);
220-
MUST_TRUE(port != 8080, "port need not equal 8080");
220+
MUST_TRUE(port != 19981, "port need not equal 19981");
221221
DEBUG(fd.fd() << " accept " << ip << ":" << port);
222222

223223
} else if (event == TcpServer::kEventRead) {
@@ -314,7 +314,7 @@ TEST(TcpServer, MixMode) {
314314
string ip;
315315
uint16_t port = 0;
316316
addr_cli.GetIPAndPort(ip, port);
317-
MUST_TRUE(port != 8080, "port need not equal 8080");
317+
MUST_TRUE(port != 19981, "port need not equal 19981");
318318
// DEBUG(fd.fd() << " accept " << ip << ":" << port);
319319
// accept
320320
} else if (event == TcpServer::kEventRead) {

‎src/test/socket/socket_test.hpp‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,7 @@
77
using namespace cppnet;
88

99
TEST(Socket, UDP) {
10-
Address addr{"127.0.0.1", 8080};
10+
Address addr{"127.0.0.1", 19981};
1111
int rc = 0;
1212

1313
Socket socket_recv, socket_send;
@@ -43,7 +43,7 @@ TEST(Socket, UDP) {
4343
}
4444

4545
TEST(Socket, CompleteRead) {
46-
Address addr{"127.0.0.1", 8080};
46+
Address addr{"127.0.0.1", 19981};
4747
int rc = 0;
4848
const int max_connect_queue = 10;
4949
std::string msg = "hello world";
@@ -123,7 +123,7 @@ TEST(Socket, ReadTimeout) {
123123
SKIP();
124124
#endif
125125

126-
Address addr{"127.0.0.1", 8080};
126+
Address addr{"127.0.0.1", 19981};
127127
int rc = 0;
128128
const int max_connect_queue = 10;
129129
std::string msg = "hello world";
@@ -154,7 +154,7 @@ TEST(Socket, ReadTimeout) {
154154
}
155155

156156
TEST(Socket, ReadUntil) {
157-
Address addr{"127.0.0.1", 8080};
157+
Address addr{"127.0.0.1", 19981};
158158
int rc = 0;
159159
const int max_connect_queue = 10;
160160
std::string msg = "hello world,cpp";

‎src/test/ssl/ssl_context_test.hpp‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,7 @@ using namespace cppnet;
1515

1616
TEST(SSLContext, SSLConnect) {
1717
SSLContext ssl_ctx;
18-
Address addr{"127.0.0.1", 8080};
18+
Address addr{"127.0.0.1", 19981};
1919
auto rc = ssl_ctx.InitSvrFile("./test/ssl/cert_demo.pem",
2020
"./test/ssl/cert_key.pem");
2121
MUST_TRUE(rc == 0, ssl_ctx.err_msg());
@@ -88,7 +88,7 @@ TEST(SSLContext, SSLConnectData) {
8888
}
8989

9090
TEST(SSLContext, SSLServer) {
91-
Address addr{"127.0.0.1", 8080};
91+
Address addr{"127.0.0.1", 19981};
9292
string msg = "hello";
9393

9494
TcpServer server(addr);

‎src/test/timer/timer_test.hpp‎

Lines changed: 78 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@ TEST(Timer, CreateTimer) {
1212
#endif
1313

1414
atomic<int> count{0};
15-
Address addr{"127.0.0.1", 8080};
15+
Address addr{"127.0.0.1", 19981};
1616
TcpServer server{addr};
1717
auto rc = server.Init();
1818
MUST_TRUE(rc == 0, server.err_msg());
@@ -62,3 +62,80 @@ TEST(Timer, CreateTimer) {
6262
"timer event count wrong " + to_string(count));
6363
#endif
6464
}
65+
66+
TEST(Timer, ResetInPlace) {
67+
#ifdef WIN32
68+
SKIP();
69+
#endif
70+
71+
// Verifies that Reset() updates the timer interval *in place* without
72+
// recreating the underlying fd. If a regression makes Reset() recreate
73+
// the fd, the new fd is not registered in epoll/kqueue and count_after
74+
// would stay zero.
75+
atomic<int> count_before{0};
76+
atomic<int> count_after{0};
77+
atomic<bool> reset_done{false};
78+
79+
Address addr{"127.0.0.1", 19981};
80+
TcpServer server{addr};
81+
auto rc = server.Init();
82+
MUST_TRUE(rc == 0, server.err_msg());
83+
84+
TimerSocket timerfd(0, 1e6); // 1ms
85+
MUST_TRUE(timerfd.status() != cppnet::Socket::kUninit, "create timerfd");
86+
DEFER_DEFAULT { timerfd.Close(); };
87+
88+
int original_fd = timerfd.fd();
89+
90+
rc = server.AddSoc(timerfd);
91+
MUST_TRUE(rc == 0, server.err_msg());
92+
93+
server.Register([&](TcpServer::Event event, TcpServer &, Socket fd) {
94+
if (event == TcpServer::kEventRead) {
95+
uint64_t exp;
96+
TimerSocket(fd).Read(&exp, sizeof(uint64_t));
97+
if (reset_done.load()) {
98+
count_after.fetch_add(1);
99+
} else {
100+
count_before.fetch_add(1);
101+
}
102+
}
103+
});
104+
105+
GO([&]() {
106+
usleep(2e4); // collect events with 1ms interval for ~20ms
107+
timerfd.Reset(0, 2e6); // slow down to 2ms
108+
MUST_TRUE(timerfd.fd() == original_fd,
109+
"Reset must not recreate fd: got " + to_string(timerfd.fd()) +
110+
", expected " + to_string(original_fd));
111+
reset_done.store(true);
112+
usleep(2e4); // observe events under the new interval for ~20ms
113+
server.Stop();
114+
});
115+
116+
rc = server.EventLoop();
117+
MUST_TRUE(rc == 0, server.err_msg());
118+
119+
DEBUG("ResetInPlace: before=" << count_before << " after=" << count_after);
120+
MUST_TRUE(count_before > 0,
121+
"no events before Reset, count=" + to_string(count_before));
122+
MUST_TRUE(count_after > 0,
123+
"no events after Reset (fd likely recreated and dropped from "
124+
"io_multiplexing), count=" +
125+
to_string(count_after));
126+
}
127+
128+
TEST(Timer, CloseIdempotent) {
129+
#ifdef WIN32
130+
SKIP();
131+
#endif
132+
133+
// Multiple Close() calls must not crash, double-free or leave dangling
134+
// resources. Memory issues here are caught by CI's valgrind run.
135+
TimerSocket timerfd(0, 1e6);
136+
MUST_TRUE(timerfd.status() != cppnet::Socket::kUninit, "create timerfd");
137+
138+
timerfd.Close();
139+
timerfd.Close();
140+
timerfd.Close();
141+
}

0 commit comments

Comments
 (0)