Fix time source mismatch in client RPC path after #3268 migration

PR #3268 ("Use monotonic time instead of wall time") migrated several
hot paths from butil::gettimeofday_us() to butil::cpuwide_time_us(),
including LocalityAwareLoadBalancer::Weight::Update which now reads
end_time_us from the cpuwide clock. The migration was incomplete on
the caller side: Channel::CallMethod and the retry sites in
Controller::OnVersionedRPCReturned still passed gettimeofday_us() as
the begin time, which is then plumbed through Controller::IssueRPC ->
Controller::Call::begin_time_us -> SelectIn::begin_time_us ->
CallInfo::begin_time_us into Weight::Update.

In Weight::Update the latency is computed as

    latency = end_time_us - ci.begin_time_us
            = cpuwide_now - wallclock_begin
            ~= -1.7e15 us (huge negative)

triggering the

    if (latency <= 0) { /* time skews, ignore the sample */ return 0; }

short-circuit. _time_q never accumulates samples, _avg_latency stays
at 0, and locality-aware feedback is silently disabled. The dual time
domain also leaks into Controller::latency_us() (controller.h:215)
where _begin_time_us was wallclock and the fallback used cpuwide.

Visible downstream effect: cold-start `list://` channels with `lb=la`
and 2 backends occasionally fail RPCs with EHOSTDOWN (`Fail to select
server from list://...`) on retry even when one backend is healthy.
Bisected reproduction in xsky/brpc fork:

  - 51 commit range c41e838..604dad0c (1.16.1 .. 1.17.0-rc2)
  - master code + LA-driven multipath probe at 2 backends, max_retry=1,
    repeat 500x:
      * commit 771de31 (one before #3268): 0/500 fail
      * commit 12fb539 (#3268):           25/500 fail
      * commit 12fb539 + revert only Weight::Update::end_time_us back
        to gettimeofday: 0/500 fail
  - Reverting the single end_time_us assignment confirms the time
    source mismatch is the regression vector.

This commit aligns the client-side callers with #3268's intent, so
all RPC time accounting uses cpuwide_time_us consistently:

    src/brpc/channel.cpp:451   CallMethod entry begin time
    src/brpc/channel.cpp:628   sync RPC end time
    src/brpc/controller.cpp:672 backup-request retry IssueRPC
    src/brpc/controller.cpp:705 retry-backoff deadline check
    src/brpc/controller.cpp:715 regular retry IssueRPC
    src/brpc/controller.cpp:803 circuit-breaker feedback latency
    src/brpc/controller.cpp:980 async OnRPCEnd
    src/brpc/controller.cpp:1019 backup-thread OnRPCEnd

Adds test/brpc_load_balancer_unittest.cpp::la_records_latency_with_consistent_time_source
which documents the LB-side invariant: when CallInfo::begin_time_us is
in the cpuwide_time domain (matching Weight::Update::end_time_us) the
LA feedback path produces a positive _avg_latency; when it is in the
wallclock domain the time-skew short-circuit fires and _avg_latency
stays 0.

Author: huangjun

src/brpc/channel.cpp

@@ -448,7 +448,7 @@ void Channel::CallMethod(const google::protobuf::MethodDescriptor* method,
                          const google::protobuf::Message* request,
                          google::protobuf::Message* response,
                          google::protobuf::Closure* done) {
-    const int64_t start_send_real_us = butil::gettimeofday_us();
+    const int64_t start_send_real_us = butil::cpuwide_time_us();
     Controller* cntl = static_cast<Controller*>(controller_base);
     cntl->OnRPCBegin(start_send_real_us);
     // Override max_retry first to reset the range of correlation_id
@@ -625,7 +625,7 @@ void Channel::CallMethod(const google::protobuf::MethodDescriptor* method,
         // fails after retry)
         Join(correlation_id);
         cntl->SubmitSpan();
-        cntl->OnRPCEnd(butil::gettimeofday_us());
+        cntl->OnRPCEnd(butil::cpuwide_time_us());
     }
 }
 

src/brpc/controller.cpp

@@ -669,7 +669,7 @@ void Controller::OnVersionedRPCReturned(const CompletionInfo& info,
         }
         ++_current_call.nretry;
         add_flag(FLAGS_BACKUP_REQUEST);
-        return IssueRPC(butil::gettimeofday_us());
+        return IssueRPC(butil::cpuwide_time_us());
     } else {
         auto retry_policy = _retry_policy ? _retry_policy : DefaultRetryPolicy();
         if (retry_policy->DoRetry(this)) {
@@ -702,7 +702,7 @@ void Controller::OnVersionedRPCReturned(const CompletionInfo& info,
             bthread::TaskGroup* g = bthread::tls_task_group;
             int64_t backoff_time_us = retry_policy->GetBackoffTimeMs(this) * 1000L;
             if (backoff_time_us > 0 &&
-                backoff_time_us < _deadline_us - butil::gettimeofday_us()) {
+                backoff_time_us < _deadline_us - butil::cpuwide_time_us()) {
                 // No need to do retry backoff when the backoff time is longer than the remaining rpc time.
                 if (retry_policy->CanRetryBackoffInPthread() ||
                     (g && !g->is_current_pthread_task())) {
@@ -712,7 +712,7 @@ void Controller::OnVersionedRPCReturned(const CompletionInfo& info,
                                     "skip retry backoff in pthread.";
                 }
             }
-            return IssueRPC(butil::gettimeofday_us());
+            return IssueRPC(butil::cpuwide_time_us());
         }
     }
 
@@ -800,7 +800,7 @@ void Controller::Call::OnComplete(
 
         if (enable_circuit_breaker) {
             sending_sock->FeedbackCircuitBreaker(error_code,
-                butil::gettimeofday_us() - begin_time_us);
+                butil::cpuwide_time_us() - begin_time_us);
         }
     }
 
@@ -977,7 +977,7 @@ void Controller::EndRPC(const CompletionInfo& info) {
             // Join is not signalled when the done does not Run() and the done
             // can't Run() because all backup threads are blocked by Join().
 
-            OnRPCEnd(butil::gettimeofday_us());
+            OnRPCEnd(butil::cpuwide_time_us());
             const bool destroy_cid_in_done = has_flag(FLAGS_DESTROY_CID_IN_DONE);
             _done->Run();
             // NOTE: Don't touch this Controller anymore, because it's likely to be
@@ -1016,7 +1016,7 @@ void Controller::RunDoneInBackupThread(void* arg) {
 void Controller::DoneInBackupThread() {
     // OnRPCEnd for sync RPC is called in Channel::CallMethod to count in
     // latency of the context-switch.
-    OnRPCEnd(butil::gettimeofday_us());
+    OnRPCEnd(butil::cpuwide_time_us());
     const CallId saved_cid = _correlation_id;
     const bool destroy_cid_in_done = has_flag(FLAGS_DESTROY_CID_IN_DONE);
     _done->Run();

test/brpc_load_balancer_unittest.cpp

@@ -1303,6 +1303,84 @@ TEST_F(LoadBalancerTest, revived_from_all_failed_intergrated) {
 }
 #endif // BUTIL_USE_ASAN
 
+// Regression for #3268's incomplete migration. Weight::Update reads
+// `end_time_us = butil::cpuwide_time_us()`, so every caller that supplies a
+// `CallInfo::begin_time_us` MUST also be in the cpuwide_time domain.
+// Before this fix, `Channel::CallMethod` (and the retry sites in
+// `Controller::OnVersionedRPCReturned`) supplied `butil::gettimeofday_us()`,
+// making `latency = end - begin` a huge negative number. Weight::Update then
+// hit the `if (latency <= 0) { /* time skews, ignore */ return 0; }`
+// short-circuit, _time_q never accumulated samples, _avg_latency stayed at 0
+// and the locality-aware feedback was effectively disabled. Visible
+// downstream symptom: cold-start `list://` channels with `lb=la` and 2
+// backends occasionally fail RPCs with `EHOSTDOWN` (Fail to select server)
+// on retry even when one backend is healthy.
+//
+// This test runs purely against LocalityAwareLoadBalancer (no Server / Channel
+// is involved), so it is hermetic. It exercises only the invariant that
+// Weight::Update relies on: matching cpuwide begin/end produces a positive
+// latency sample.
+TEST_F(LoadBalancerTest, la_records_latency_with_consistent_time_source) {
+    LALB lalb;
+    char addr[] = "192.168.1.1:8080";
+    butil::EndPoint dummy;
+    ASSERT_EQ(0, str2endpoint(addr, &dummy));
+    brpc::ServerId id(8888);
+    brpc::SocketOptions options;
+    options.remote_side = dummy;
+    ASSERT_EQ(0, brpc::Socket::Create(options, &id.id));
+    ASSERT_TRUE(lalb.AddServer(id));
+
+    auto avg_latency = [&]() -> int64_t {
+        std::ostringstream os;
+        brpc::DescribeOptions opts;
+        opts.verbose = true;
+        lalb.Describe(os, opts);
+        const std::string s = os.str();
+        const size_t p = s.find("avg_latency=");
+        if (p == std::string::npos) return -1;
+        return strtoll(s.c_str() + p + strlen("avg_latency="), NULL, 10);
+    };
+
+    // Path A (regression repro): caller supplies wall-clock begin_time_us.
+    // Weight::Update sees `latency = cpuwide_now - wallclock_begin = HUGE
+    // NEGATIVE`, drops the sample. _avg_latency stays at 0.
+    for (int i = 0; i < 8; ++i) {
+        const int64_t bad_begin = butil::gettimeofday_us();
+        brpc::SocketUniquePtr ptr;
+        brpc::LoadBalancer::SelectIn in = { bad_begin, true, false, 0u, NULL };
+        brpc::LoadBalancer::SelectOut out(&ptr);
+        ASSERT_EQ(0, lalb.SelectServer(in, &out));
+        bthread_usleep(2000);
+        brpc::LoadBalancer::CallInfo ci = { bad_begin, id.id, 0, NULL };
+        lalb.Feedback(ci);
+    }
+    EXPECT_EQ(0, avg_latency())
+        << "Wall-clock begin_time_us must trip the time-skew short-circuit; "
+        << "if avg_latency is non-zero here, either Weight::Update changed "
+        << "its time source or the loop above is misconfigured.";
+
+    // Path B (post-fix): caller supplies cpuwide begin_time_us, matching
+    // Weight::Update::end_time_us. _avg_latency now reflects the actual
+    // elapsed latency (~2ms = 2000us).
+    for (int i = 0; i < 8; ++i) {
+        const int64_t good_begin = butil::cpuwide_time_us();
+        brpc::SocketUniquePtr ptr;
+        brpc::LoadBalancer::SelectIn in = { good_begin, true, false, 0u, NULL };
+        brpc::LoadBalancer::SelectOut out(&ptr);
+        ASSERT_EQ(0, lalb.SelectServer(in, &out));
+        bthread_usleep(2000);
+        brpc::LoadBalancer::CallInfo ci = { good_begin, id.id, 0, NULL };
+        lalb.Feedback(ci);
+    }
+    EXPECT_GT(avg_latency(), 0)
+        << "cpuwide begin_time_us should yield positive latency feedback. "
+        << "If this is 0, callers in Channel/Controller may have regressed "
+        << "back to gettimeofday_us() for begin_time_us.";
+
+    ASSERT_EQ(0, brpc::Socket::SetFailed(id.id));
+}
+
 TEST_F(LoadBalancerTest, la_selection_too_long) {
     brpc::GlobalInitializeOrDie();
     brpc::LoadBalancerWithNaming lb;