Only update real sent time to span

chenBright · chenBright · commit 86f6795cfef1 · 2025-02-09T15:08:10.000+08:00
diff --git a/.github/actions/init-ut-make-config/action.yml b/.github/actions/init-ut-make-config/action.yml
@@ -9,5 +9,5 @@ runs:
       shell: bash
     - run: sudo apt-get update && sudo apt-get install -y libgtest-dev cmake gdb libstdc++6-9-dbg && cd /usr/src/gtest && sudo cmake . && sudo make -j ${{env.proc_num}} && sudo mv lib/libgtest* /usr/lib/
       shell: bash
-    - run: sh config_brpc.sh --headers="/libunwind/include /usr/include" --libs="/libunwind/lib /usr/lib /usr/lib64" --nodebugsymbols ${{inputs.options}}
+    - run: sh config_brpc.sh --headers="/libunwind/include /usr/include" --libs="/libunwind/lib /usr/lib /usr/lib64" ${{inputs.options}}
       shell: bash
diff --git a/src/brpc/details/method_status.cpp b/src/brpc/details/method_status.cpp
@@ -149,22 +149,16 @@ void MethodStatus::SetConcurrencyLimiter(ConcurrencyLimiter* cl) {
     _cl.reset(cl);
 }
 
-int HandleResponseWritten(bthread_id_t id, void* data, int error_code,
-    const std::string& error_text) {
+int HandleResponseWritten(bthread_id_t id, void* data, int /*error_code*/) {
     auto args = static_cast<ResponseWriteInfo*>(data);
-    args->error_code = error_code;
-    args->error_text = error_text;
     args->sent_us = butil::cpuwide_time_us();
     CHECK_EQ(0, bthread_id_unlock_and_destroy(id));
     return 0;
 }
 
 ConcurrencyRemover::~ConcurrencyRemover() {
     if (_status) {
-        if (_sent_us < _received_us) {
-            _sent_us = butil::cpuwide_time_us();
-        }
-        _status->OnResponded(_c->ErrorCode(), _sent_us - _received_us);
+        _status->OnResponded(_c->ErrorCode(), butil::cpuwide_time_us() - _received_us);
         _status = NULL;
     }
     ServerPrivateAccessor(_c->server()).RemoveConcurrency(_c);
diff --git a/src/brpc/details/method_status.h b/src/brpc/details/method_status.h
@@ -76,27 +76,22 @@ friend class Server;
 };
 
 struct ResponseWriteInfo {
-    int error_code{0};
-    std::string error_text;
     int64_t sent_us{0};
 };
 
-int HandleResponseWritten(bthread_id_t id, void* data, int error_code,
-                          const std::string& error_text);
+int HandleResponseWritten(bthread_id_t id, void* data, int error_code);
 
 class ConcurrencyRemover {
 public:
     ConcurrencyRemover(MethodStatus* status, Controller* c, int64_t received_us)
         : _status(status) , _c(c) , _received_us(received_us) {}
     ~ConcurrencyRemover();
 
-    void set_sent_us(int64_t sent_us) { _sent_us = sent_us; }
 private:
     DISALLOW_COPY_AND_ASSIGN(ConcurrencyRemover);
     MethodStatus* _status;
     Controller* _c;
     int64_t _received_us;
-    int64_t _sent_us{0};
 };
 
 inline bool MethodStatus::OnRequested(int* rejected_cc, Controller* cntl) {
diff --git a/src/brpc/policy/baidu_rpc_protocol.cpp b/src/brpc/policy/baidu_rpc_protocol.cpp
@@ -197,19 +197,15 @@ void SendRpcResponse(int64_t correlation_id, Controller* cntl,
     const google::protobuf::Message* req = NULL == messages ? NULL : messages->Request();
     const google::protobuf::Message* res = NULL == messages ? NULL : messages->Response();
 
-    ResponseWriteInfo args;
-
     // Recycle resources at the end of this function.
     BRPC_SCOPE_EXIT {
         {
             // Remove concurrency and record latency at first.
             ConcurrencyRemover concurrency_remover(method_status, cntl, received_us);
-            concurrency_remover.set_sent_us(args.sent_us);
         }
 
         std::unique_ptr<Controller, LogErrorTextAndDelete> recycle_cntl(cntl);
 
-
         if (NULL == messages) {
             return;
         }
@@ -305,12 +301,13 @@ void SendRpcResponse(int64_t correlation_id, Controller* cntl,
         }
     }
 
+    ResponseWriteInfo args;
+    bthread_id_t response_id = INVALID_BTHREAD_ID;
     if (span) {
         span->set_response_size(res_buf.size());
+        CHECK_EQ(0, bthread_id_create(&response_id, &args, HandleResponseWritten));
     }
 
-    bthread_id_t response_id;
-    CHECK_EQ(0, bthread_id_create2(&response_id, &args, HandleResponseWritten));
     // Send rpc response over stream even if server side failed to create
     // stream for some reason.
     if (cntl->has_remote_stream()) {
@@ -328,45 +325,13 @@ void SendRpcResponse(int64_t correlation_id, Controller* cntl,
             Stream::SetFailed(response_stream_ids, error_code,
                               "Fail to write into %s",
                               sock->description().c_str());
-        }
-    } else{
-        // Have the risk of unlimited pending responses, in which case, tell
-        // users to set max_concurrency.
-        Socket::WriteOptions wopt;
-        wopt.id_wait = response_id;
-        wopt.notify_on_success = true;
-        wopt.ignore_eovercrowded = true;
-        if (sock->Write(&res_buf, &wopt) != 0) {
-            const int errcode = errno;
-            PLOG_IF(WARNING, errcode != EPIPE) << "Fail to write into " << *sock;
-            cntl->SetFailed(errcode, "Fail to write into %s",
-                            sock->description().c_str());
-            return;
-        }
-    }
-
-    bthread_id_join(response_id);
-
-    error_code = args.error_code;
-    if (cntl->has_remote_stream()) {
-        if (0 != error_code) {
-            LOG_IF(WARNING, error_code != EPIPE)
-                << "Fail to write into " << *sock
-                << ", error text= " << args.error_text
-                << ": " << berror(error_code);
-            cntl->SetFailed(error_code,  "Fail to write into %s: %s",
-                            sock->description().c_str(),
-                            args.error_text.c_str());
-            Stream::SetFailed(response_stream_ids, error_code,
-                              "Fail to write into %s",
-                              args.error_text.c_str());
             return;
         }
 
         // Now it's ok the mark these server-side streams as connected as all the
         // written user data would follower the RPC response.
         // Reuse stream_ptr to avoid address first stream id again
-        if (NULL != stream_ptr) {
+        if (stream_ptr) {
             ((Stream*)stream_ptr->conn())->SetConnected();
         }
         for (size_t i = 1; i < response_stream_ids.size(); ++i) {
@@ -384,17 +349,24 @@ void SendRpcResponse(int64_t correlation_id, Controller* cntl,
     } else{
         // Have the risk of unlimited pending responses, in which case, tell
         // users to set max_concurrency.
-        if (0 != error_code) {
-            LOG_IF(WARNING, error_code != EPIPE) << "Fail to write into " << *sock
-                                              << ", error text= " << args.error_text
-                                              << ": " << strerror(error_code);
-            cntl->SetFailed(error_code,  "Fail to write into %s: %s",
-                            sock->description().c_str(), args.error_text.c_str());
+        Socket::WriteOptions wopt;
+        wopt.ignore_eovercrowded = true;
+        if (INVALID_BTHREAD_ID != response_id) {
+            wopt.id_wait = response_id;
+            wopt.notify_on_success = true;
+        }
+        if (sock->Write(&res_buf, &wopt) != 0) {
+            const int errcode = errno;
+            PLOG_IF(WARNING, errcode != EPIPE) << "Fail to write into " << *sock;
+            cntl->SetFailed(errcode, "Fail to write into %s",
+                            sock->description().c_str());
             return;
         }
     }
 
     if (span) {
+        bthread_id_join(response_id);
+        // Do not care about the result of background writing.
         // TODO: this is not sent
         span->set_sent_us(args.sent_us);
     }
diff --git a/src/brpc/policy/http_rpc_protocol.cpp b/src/brpc/policy/http_rpc_protocol.cpp
@@ -935,12 +935,14 @@ HttpResponseSender::~HttpResponseSender() {
     // Have the risk of unlimited pending responses, in which case, tell
     // users to set max_concurrency.
     ResponseWriteInfo args;
-    bthread_id_t response_id;
-    CHECK_EQ(0, bthread_id_create2(&response_id, &args, HandleResponseWritten));
     Socket::WriteOptions wopt;
-    wopt.id_wait = response_id;
-    wopt.notify_on_success = true;
     wopt.ignore_eovercrowded = true;
+    bthread_id_t response_id = INVALID_BTHREAD_ID;
+    if (span) {
+        CHECK_EQ(0, bthread_id_create(&response_id, &args, HandleResponseWritten));
+        wopt.id_wait = response_id;
+        wopt.notify_on_success = true;
+    }
     if (is_http2) {
         if (is_grpc) {
             // Append compressed and length before body
@@ -986,16 +988,9 @@ HttpResponseSender::~HttpResponseSender() {
         return;
     }
 
-    bthread_id_join(response_id);
-    concurrency_remover.set_sent_us(args.sent_us);
-    const int errcode = args.error_code;
-    if (0 != errcode) {
-        LOG_IF(WARNING, errcode != EPIPE) << "Fail to write into " << *socket;
-        cntl->SetFailed(errcode, "Fail to write into %s", socket->description().c_str());
-        return;
-    }
-
     if (span) {
+        bthread_id_join(response_id);
+        // Do not care about the result of background writing.
         // TODO: this is not sent
         span->set_sent_us(args.sent_us);
     }
diff --git a/src/brpc/policy/hulu_pbrpc_protocol.cpp b/src/brpc/policy/hulu_pbrpc_protocol.cpp
@@ -305,12 +305,14 @@ static void SendHuluResponse(int64_t correlation_id,
     // Have the risk of unlimited pending responses, in which case, tell
     // users to set max_concurrency.
     ResponseWriteInfo args;
-    bthread_id_t response_id;
-    CHECK_EQ(0, bthread_id_create2(&response_id, &args, HandleResponseWritten));
     Socket::WriteOptions wopt;
-    wopt.id_wait = response_id;
-    wopt.notify_on_success = true;
     wopt.ignore_eovercrowded = true;
+    bthread_id_t response_id = INVALID_BTHREAD_ID;
+    if (span) {
+        CHECK_EQ(0, bthread_id_create(&response_id, &args, HandleResponseWritten));
+        wopt.id_wait = response_id;
+        wopt.notify_on_success = true;
+    }
     if (sock->Write(&res_buf, &wopt) != 0) {
         const int errcode = errno;
         PLOG_IF(WARNING, errcode != EPIPE) << "Fail to write into " << *sock;
@@ -319,16 +321,9 @@ static void SendHuluResponse(int64_t correlation_id,
         return;
     }
 
-    bthread_id_join(response_id);
-    concurrency_remover.set_sent_us(args.sent_us);
-    const int errcode = args.error_code;
-    if (0 != errcode) {
-        LOG_IF(WARNING, errcode != EPIPE) << "Fail to write into " << *sock;
-        cntl->SetFailed(errcode, "Fail to write into %s", sock->description().c_str());
-        return;
-    }
-
     if (span) {
+        bthread_id_join(response_id);
+        // Do not care about the result of background writing.
         // TODO: this is not sent
         span->set_sent_us(args.sent_us);
     }
diff --git a/src/brpc/policy/mongo_protocol.cpp b/src/brpc/policy/mongo_protocol.cpp
@@ -95,34 +95,16 @@ void SendMongoResponse::Run() {
         res_buf.append(res.message());
     }
 
-    if (res_buf.empty()) {
-        return;
-    }
-
-    // Have the risk of unlimited pending responses, in which case, tell
-    // users to set max_concurrency.
-    ResponseWriteInfo args;
-    bthread_id_t response_id;
-    CHECK_EQ(0, bthread_id_create2(&response_id, &args, HandleResponseWritten));
-    Socket::WriteOptions wopt;
-    wopt.id_wait = response_id;
-    wopt.notify_on_success = true;
-    wopt.ignore_eovercrowded = true;
-    wopt.ignore_eovercrowded = true;
-    if (socket->Write(&res_buf, &wopt) != 0) {
-        PLOG(WARNING) << "Fail to write into " << *socket;
-        return;
-    }
-
-    bthread_id_join(response_id);
-    concurrency_remover.set_sent_us(args.sent_us);
-    const int errcode = args.error_code;
-    if (0 != errcode) {
-        LOG_IF(WARNING, errcode != EPIPE) << "Fail to write into " << *socket;
-        cntl.SetFailed(errcode, "Fail to write into %s", socket->description().c_str());
-        return;
+    if (!res_buf.empty()) {
+        // Have the risk of unlimited pending responses, in which case, tell
+        // users to set max_concurrency.
+        Socket::WriteOptions wopt;
+        wopt.ignore_eovercrowded = true;
+        if (socket->Write(&res_buf, &wopt) != 0) {
+            PLOG(WARNING) << "Fail to write into " << *socket;
+            return;
+        }
     }
-
 }
 
 ParseResult ParseMongoMessage(butil::IOBuf* source,
diff --git a/src/brpc/policy/nshead_protocol.cpp b/src/brpc/policy/nshead_protocol.cpp
@@ -114,12 +114,14 @@ void NsheadClosure::Run() {
         // Have the risk of unlimited pending responses, in which case, tell
         // users to set max_concurrency.
         ResponseWriteInfo args;
-        bthread_id_t response_id;
-        CHECK_EQ(0, bthread_id_create2(&response_id, &args, HandleResponseWritten));
         Socket::WriteOptions wopt;
-        wopt.id_wait = response_id;
-        wopt.notify_on_success = true;
         wopt.ignore_eovercrowded = true;
+        bthread_id_t response_id = INVALID_BTHREAD_ID;
+        if (span) {
+            CHECK_EQ(0, bthread_id_create(&response_id, &args, HandleResponseWritten));
+            wopt.id_wait = response_id;
+            wopt.notify_on_success = true;
+        }
         if (sock->Write(&write_buf, &wopt) != 0) {
             const int errcode = errno;
             PLOG_IF(WARNING, errcode != EPIPE) << "Fail to write into " << *sock;
@@ -128,16 +130,11 @@ void NsheadClosure::Run() {
             return;
         }
 
-        bthread_id_join(response_id);
-        concurrency_remover.set_sent_us(args.sent_us);
-        const int errcode = args.error_code;
-        if (0 != errcode) {
-            LOG_IF(WARNING, errcode != EPIPE) << "Fail to write into " << *sock;
-            _controller.SetFailed(errcode, "Fail to write into %s",
-                                  sock->description().c_str());
-            return;
+        if (span) {
+            bthread_id_join(response_id);
+            // Do not care about the result of background writing.
+            sent_us = args.sent_us;
         }
-        sent_us = args.sent_us;
     }
     if (span) {
         // TODO: this is not sent
diff --git a/src/brpc/policy/sofa_pbrpc_protocol.cpp b/src/brpc/policy/sofa_pbrpc_protocol.cpp
@@ -282,12 +282,14 @@ static void SendSofaResponse(int64_t correlation_id,
     // Have the risk of unlimited pending responses, in which case, tell
     // users to set max_concurrency.
     ResponseWriteInfo args;
-    bthread_id_t response_id;
-    CHECK_EQ(0, bthread_id_create2(&response_id, &args, HandleResponseWritten));
     Socket::WriteOptions wopt;
-    wopt.id_wait = response_id;
-    wopt.notify_on_success = true;
     wopt.ignore_eovercrowded = true;
+    bthread_id_t response_id = INVALID_BTHREAD_ID;
+    if (span) {
+        CHECK_EQ(0, bthread_id_create(&response_id, &args, HandleResponseWritten));
+        wopt.id_wait = response_id;
+        wopt.notify_on_success = true;
+    }
     if (sock->Write(&res_buf, &wopt) != 0) {
         const int errcode = errno;
         PLOG_IF(WARNING, errcode != EPIPE) << "Fail to write into " << *sock;
@@ -296,17 +298,9 @@ static void SendSofaResponse(int64_t correlation_id,
         return;
     }
 
-    bthread_id_join(response_id);
-    concurrency_remover.set_sent_us(args.sent_us);
-    const int errcode = args.error_code;
-    if (0 != errcode) {
-        LOG_IF(WARNING, errcode != EPIPE) << "Fail to write into " << *sock;
-        cntl->SetFailed(errcode, "Fail to write into %s",
-                        sock->description().c_str());
-        return;
-    }
-
     if (span) {
+        bthread_id_join(response_id);
+        // Do not care about the result of background writing.
         // TODO: this is not sent
         span->set_sent_us(args.sent_us);
     }
diff --git a/src/brpc/policy/streaming_rpc_protocol.cpp b/src/brpc/policy/streaming_rpc_protocol.cpp
@@ -154,8 +154,7 @@ void SendStreamClose(Socket *sock, int64_t remote_stream_id,
 }
 
 int SendStreamData(Socket* sock, const butil::IOBuf* data,
-                   int64_t remote_stream_id,
-                   int64_t source_stream_id,
+                   int64_t remote_stream_id, int64_t source_stream_id,
                    bthread_id_t response_id) {
     CHECK(sock != NULL);
     StreamFrameMeta fm;
@@ -166,8 +165,10 @@ int SendStreamData(Socket* sock, const butil::IOBuf* data,
     butil::IOBuf out;
     PackStreamMessage(&out, fm, data);
     Socket::WriteOptions wopt;
-    wopt.id_wait = response_id;
-    wopt.notify_on_success = true;
+    if (INVALID_BTHREAD_ID != response_id) {
+        wopt.id_wait = response_id;
+        wopt.notify_on_success = true;
+    }
     wopt.ignore_eovercrowded = true;
     return sock->Write(&out, &wopt);
 }
diff --git a/src/brpc/policy/streaming_rpc_protocol.h b/src/brpc/policy/streaming_rpc_protocol.h
