Fix bugs in calculating latency and qps frequency inaccuracies

example/auto_concurrency_limiter/client.cpp

@@ -31,9 +31,10 @@ DEFINE_string(cntl_server, "0.0.0.0:9000", "IP Address of server");
 DEFINE_string(echo_server, "0.0.0.0:9001", "IP Address of server");
 DEFINE_int32(timeout_ms, 3000, "RPC timeout in milliseconds");
 DEFINE_int32(max_retry, 0, "Max retries(not including the first RPC)"); 
-DEFINE_int32(case_interval, 20, "");
-DEFINE_int32(client_frequent_interval_us, 10000, "");
-DEFINE_string(case_file, "", "");
+DEFINE_int32(case_interval, 20, "Intervals for different test cases");
+DEFINE_int32(client_qps_change_interval_us, 50000, 
+    "The interval for client changes the sending speed");
+DEFINE_string(case_file, "", "File path for test_cases");
 
 void DisplayStage(const test::Stage& stage) {
     std::string type;
@@ -156,10 +157,9 @@ struct TestCaseContext {
 void RunUpdateTask(void* data) {
     TestCaseContext* context = (TestCaseContext*)data;
     bool should_continue = context->Update();
-    timespec ts;
-    ts.tv_nsec = FLAGS_client_frequent_interval_us * 1000;
     if (should_continue) {
-        bthread::get_global_timer_thread()->schedule(RunUpdateTask, data, ts);
+        bthread::get_global_timer_thread()->schedule(RunUpdateTask, data, 
+            butil::microseconds_from_now(FLAGS_client_qps_change_interval_us));
     } else {
         context->running.store(false, butil::memory_order_release);
     }
@@ -173,6 +173,7 @@ void RunCase(test::ControlService_Stub &cntl_stub,
     options.protocol = FLAGS_protocol;
     options.connection_type = FLAGS_connection_type;
     options.timeout_ms = FLAGS_timeout_ms;
+    options.max_retry = FLAGS_max_retry;
     if (channel.Init(FLAGS_echo_server.c_str(), &options) != 0) {
         LOG(FATAL) << "Fail to initialize channel";
     }
@@ -186,9 +187,8 @@ void RunCase(test::ControlService_Stub &cntl_stub,
     CHECK(!cntl.Failed()) << "control failed";
 
     TestCaseContext context(test_case);
-    timespec ts;
-    ts.tv_nsec = FLAGS_client_frequent_interval_us * 1000;
-    bthread::get_global_timer_thread()->schedule(RunUpdateTask, &context, ts);
+    bthread::get_global_timer_thread()->schedule(RunUpdateTask, &context, 
+        butil::microseconds_from_now(FLAGS_client_qps_change_interval_us));
 
     while (context.running.load(butil::memory_order_acquire)) {
         test::NotifyRequest echo_req;

example/auto_concurrency_limiter/server.cpp

@@ -31,14 +31,18 @@
 DEFINE_int32(logoff_ms, 2000, "Maximum duration of server's LOGOFF state "
              "(waiting for client to close connection before server stops)");
 DEFINE_int32(server_bthread_concurrency, 4, "For compute max qps");
-DEFINE_int32(server_sync_sleep_us, 2500, "For compute max qps");
+DEFINE_int32(server_sync_sleep_us, 2500, "For compute maximum qps");
 // max qps = 1000 / 2.5 * 4 = 1600
 
 DEFINE_int32(control_server_port, 9000, "");
-DEFINE_int32(echo_port, 9001, "");
-DEFINE_int32(cntl_port, 9000, "TCP Port of this server");
-DEFINE_string(case_file, "", "");
-DEFINE_int32(server_frequent_interval_us, 10000, "");
+DEFINE_int32(echo_port, 9001, "TCP Port of echo server");
+DEFINE_int32(cntl_port, 9000, "TCP Port of controller server");
+DEFINE_string(case_file, "", "File path for test_cases");
+DEFINE_int32(latency_change_interval_us, 50000, "Intervalt for server side changes the latency");
+DEFINE_int32(server_max_concurrency, 0, "Echo Server's max_concurrency");
+DEFINE_bool(use_usleep, false, 
+        "EchoServer uses ::usleep or bthread_usleep to simulate latency "
+        "when processing requests");
 
 
 bthread::TimerThread g_timer_thread;
@@ -110,9 +114,8 @@ public:
             return;
         }
         ComputeLatency();
-        timespec ts;
-        ts.tv_nsec = FLAGS_server_frequent_interval_us * 1000;
-        g_timer_thread.schedule(TimerTask, (void*)this, ts);
+        g_timer_thread.schedule(TimerTask, (void*)this, 
+            butil::microseconds_from_now(FLAGS_latency_change_interval_us));
     }
 
     virtual void Echo(google::protobuf::RpcController* cntl_base,
@@ -122,7 +125,11 @@ public:
         brpc::ClosureGuard done_guard(done); 
         response->set_message("hello");
         ::usleep(FLAGS_server_sync_sleep_us);
-        bthread_usleep(_latency.load(butil::memory_order_relaxed));
+        if (FLAGS_use_usleep) {
+            ::usleep(_latency.load(butil::memory_order_relaxed));
+        } else {
+            bthread_usleep(_latency.load(butil::memory_order_relaxed));
+        }
     }
 
     void ComputeLatency() {
@@ -213,9 +220,11 @@ public:
             CHECK(!_server.IsRunning()) << "Continuous StartCase";
             const test::TestCase& test_case = _case_set.test_case(_case_index++);
             _echo_service->SetTestCase(test_case);
+            brpc::ServerOptions options;
+//            options.max_concurrency = 15;
             _server.MaxConcurrencyOf("test.EchoService.Echo") = test_case.max_concurrency();
 
-            _server.Start(FLAGS_echo_port, NULL);            
+            _server.Start(FLAGS_echo_port, &options);            
             _echo_service->StartTestCase();
             response->set_message("CaseStarted");
         } else if (message == "StopCase") {

example/auto_concurrency_limiter/settings.flags

---ABTest=true
---auto_cl_min_reserved_concurrency=20
---case_file=test_case.json
+--case_file=test_case_test
+--client_qps_change_interval_us=50000
+--max_retry=0
 
---client_frequent_interval_us=5000000
+--auto_cl_overload_threshold=0.3
+--auto_cl_initial_max_concurrency=16
 
---server_frequent_interval_us=5000000
+--latency_change_interval_us=50000
+--server_bthread_concurrency=4
+--server_sync_sleep_us=2500
+--use_usleep=false

example/auto_concurrency_limiter/test_case.json

@@ -8,7 +8,7 @@
         {
             "lower_bound":3000,
             "upper_bound":3000,
-            "duration_sec":3,
+            "duration_sec":30,
             "type":2
         }
     ],
@@ -17,7 +17,7 @@
         {
             "lower_bound":20000,
             "upper_bound":20000,
-            "duration_sec":3,
+            "duration_sec":30,
             "type":2
         }
     ]