Star5
66
代码
代码
Issues1
Pull Requests15
项目讨论
Wiki
分析
Star5
66
草稿
[WIP]修改工具的发包机制 #174
Jinhui Tong创建于 7月20日
[WIP]修改工具的发包机制 #174
草稿
Jinhui Tong创建于 7月20日
从tongjinhui/brpcmaster合入到fanzhaonan/brpcmaster
共 2 个文件变更+22-12
Mexample/ub_test/client.cpp+21-11
@@ -83,6 +83,7 @@ DEFINE_int32(token_interval_period, 1, "Token generation interval period multipl
83DEFINE_bool(use_connection_group, false, "Set connection_group for each channel or not");83DEFINE_bool(use_connection_group, false, "Set connection_group for each channel or not");
84DEFINE_bool(test_keep_alive, false, "Keep connections alive for 10 seconds after establishment");84DEFINE_bool(test_keep_alive, false, "Keep connections alive for 10 seconds after establishment");
85DEFINE_bool(debug_latency, false, "Print latency of each successful request");85DEFINE_bool(debug_latency, false, "Print latency of each successful request");
86+DEFINE_int32(max_random_delay_us, 5000, "Max random delay in microseconds before sending request (used when expected_qps > 0)");
86DEFINE_string(export_percentile_file, "", "Export percentile samples to a binary file for cross-instance aggregation (empty to skip)");87DEFINE_string(export_percentile_file, "", "Export percentile samples to a binary file for cross-instance aggregation (empty to skip)");
87 88 
88// ==================== 全局变量 ====================89// ==================== 全局变量 ====================
@@ -92,8 +93,8 @@ bvar::LatencyRecorder g_server_cpu_recorder("server_cpu");
92bvar::LatencyRecorder g_client_cpu_recorder("client_cpu");93bvar::LatencyRecorder g_client_cpu_recorder("client_cpu");
93bvar::LatencyRecorder g_client_memory_recorder("client_memory");94bvar::LatencyRecorder g_client_memory_recorder("client_memory");
94 95 
95-// 令牌桶96+// 令牌桶 (使用协程信号量替代原子变量，避免忙等)
96-butil::atomic<int64_t> g_token(10000);97+bthread_sem_t g_token_sem;
97volatile bool g_stop = false;98volatile bool g_stop = false;
98 99 
99// 首包延迟存储 (需要 mutex 保护)100// 首包延迟存储 (需要 mutex 保护)
@@ -312,11 +313,9 @@ public:
312 // 发送请求313 // 发送请求
313 void SendRequest() {314 void SendRequest() {
314 if (FLAGS_expected_qps > 0) {315 if (FLAGS_expected_qps > 0) {
315- while (g_token.load(butil::memory_order_relaxed) <= 0) {316+ // 协程信号量阻塞等待，不占用 CPU
316- // bthread_usleep(2000 + butil::fast_rand_less_than(1001));317+ bthread_sem_wait(&g_token_sem);
317- bthread_usleep(2000);318+ bthread_usleep(butil::fast_rand_less_than(FLAGS_max_random_delay_us));
318- }
319- g_token.fetch_sub(1, butil::memory_order_relaxed);
320 }319 }
321 RespClosure* closure = new RespClosure;320 RespClosure* closure = new RespClosure;
322 test::PerfTestRequest request;321 test::PerfTestRequest request;
@@ -418,16 +417,20 @@ private:
418};417};
419 418 
420// ==================== 令牌桶生成器 ====================419// ==================== 令牌桶生成器 ====================
420+// 使用协程信号量实现令牌桶，通过 bthread_sem_post_n 发放令牌，
421+// 消费端通过 bthread_sem_wait 阻塞等待，避免忙等自旋。
421static void* GenerateToken(void* arg) {422static void* GenerateToken(void* arg) {
422 int64_t start_time = butil::monotonic_time_ns();423 int64_t start_time = butil::monotonic_time_ns();
423- int64_t accumulative_token = g_token.load(butil::memory_order_relaxed);424+ int64_t accumulative_token = FLAGS_initial_tokens;
424 while (!g_stop) {425 while (!g_stop) {
425 bthread_usleep(10000 * FLAGS_token_interval_period);426 bthread_usleep(10000 * FLAGS_token_interval_period);
426 int64_t now = butil::monotonic_time_ns();427 int64_t now = butil::monotonic_time_ns();
427 if (accumulative_token * 1000000000 / (now - start_time) < FLAGS_expected_qps) {428 if (accumulative_token * 1000000000 / (now - start_time) < FLAGS_expected_qps) {
428 int64_t delta = FLAGS_expected_qps * (now - start_time) / 1000000000 - accumulative_token;429 int64_t delta = FLAGS_expected_qps * (now - start_time) / 1000000000 - accumulative_token;
429- g_token.fetch_add(delta, butil::memory_order_relaxed);430+ if (delta > 0) {
430- accumulative_token += delta;431+ bthread_sem_post_n(&g_token_sem, static_cast<size_t>(delta));
432+ accumulative_token += delta;
433+ }
431 }434 }
432 }435 }
433 return NULL;436 return NULL;
@@ -672,6 +675,12 @@ void Test(int thread_num, int attachment_size) {
672 << ", QueueDepth: " << FLAGS_queue_depth675 << ", QueueDepth: " << FLAGS_queue_depth
673 << "]" << std::endl;676 << "]" << std::endl;
674 677 
678+ // 重置令牌桶信号量 (支持多次 Test() 调用时重新初始化)
679+ if (FLAGS_expected_qps > 0) {
680+ bthread_sem_destroy(&g_token_sem);
681+ bthread_sem_init(&g_token_sem, static_cast<unsigned>(FLAGS_initial_tokens));
682+ }
683+ 
675 // 重置全局计数器684 // 重置全局计数器
676 g_total_bytes.store(0, butil::memory_order_relaxed);685 g_total_bytes.store(0, butil::memory_order_relaxed);
677 g_all_cnt.store(0, butil::memory_order_relaxed);686 g_all_cnt.store(0, butil::memory_order_relaxed);
@@ -759,7 +768,8 @@ int main(int argc, char* argv[]) {
759 GFLAGS_NAMESPACE::ParseCommandLineFlags(&argc, &argv, true);768 GFLAGS_NAMESPACE::ParseCommandLineFlags(&argc, &argv, true);
760 g_name.resize(FLAGS_req_size, 'r');769 g_name.resize(FLAGS_req_size, 'r');
761 770 
762- g_token.store(FLAGS_initial_tokens);771+ // bthread_sem_init(&g_token_sem, static_cast<unsigned>(FLAGS_initial_tokens));
772+ bthread_sem_init(&g_token_sem, 0);
763 773 
764#ifdef WITH_RDMA774#ifdef WITH_RDMA
765 if (FLAGS_use_rdma) {775 if (FLAGS_use_rdma) {
Msrc/bthread/semaphore.cpp+1-1
@@ -116,7 +116,7 @@ static inline int bthread_sem_post(bthread_sem_t* sem, size_t num) {
116 bvar::is_collectable(&bthread::g_cp_sl) : bvar::INVALID_SAMPLING_RANGE;116 bvar::is_collectable(&bthread::g_cp_sl) : bvar::INVALID_SAMPLING_RANGE;
117 const int64_t start_ns = bvar::is_sampling_range_valid(sampling_range) ?117 const int64_t start_ns = bvar::is_sampling_range_valid(sampling_range) ?
118 butil::cpuwide_time_ns() : -1;118 butil::cpuwide_time_ns() : -1;
119- bthread::butex_wake_n(sem->butex, n);119+ bthread::butex_wake_n(sem->butex, num);
120 if (start_ns > 0) {120 if (start_ns > 0) {
121 const int64_t end_ns = butil::cpuwide_time_ns();121 const int64_t end_ns = butil::cpuwide_time_ns();
122 const bthread_contention_site_t csite{end_ns - start_ns, sampling_range};122 const bthread_contention_site_t csite{end_ns - start_ns, sampling_range};