This is an automated email from the ASF dual-hosted git repository.
wwbmmm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/brpc.git
The following commit(s) were added to refs/heads/master by this push:
new e0abb100 Progressive timeout dev (#3409)
e0abb100 is described below
commit e0abb1001ef76d400dddc2b2c3a1ff42f5d3afac
Author: Chuang Zhang <[email protected]>
AuthorDate: Sat Aug 15 16:56:56 2026 +0800
Progressive timeout dev (#3409)
* add the progressive timeout reader
* optimize the code format
* WIP: progressive read timeout review
* test: strengthen progressive read timeout coverage
---
example/http_c++/http_client.cpp | 35 +++++
example/http_c++/http_server.cpp | 7 +
src/brpc/controller.cpp | 259 ++++++++++++++++++++++++++++++-
src/brpc/controller.h | 8 +-
src/brpc/errno.proto | 1 +
src/brpc/policy/http_rpc_protocol.cpp | 1 +
src/brpc/policy/http_rpc_protocol.h | 12 +-
src/brpc/progressive_reader.h | 2 +
test/brpc_http_rpc_protocol_unittest.cpp | 237 +++++++++++++++++++++++++++-
9 files changed, 557 insertions(+), 5 deletions(-)
diff --git a/example/http_c++/http_client.cpp b/example/http_c++/http_client.cpp
index 23222dee..3a09186f 100644
--- a/example/http_c++/http_client.cpp
+++ b/example/http_c++/http_client.cpp
@@ -22,11 +22,15 @@
// - Access www.foo.com
// ./http_client www.foo.com
+#include <string>
#include <gflags/gflags.h>
#include <butil/logging.h>
#include <brpc/channel.h>
+#include "bthread/countdown_event.h"
DEFINE_string(d, "", "POST this data to the http server");
+DEFINE_bool(progressive, false, "whether or not progressive read data from
server");
+DEFINE_int32(progressive_read_timeout_ms, 5000, "progressive read data idle
timeout in milliseconds");
DEFINE_string(load_balancer, "", "The algorithm for load balancing");
DEFINE_int32(timeout_ms, 2000, "RPC timeout in milliseconds");
DEFINE_int32(max_retry, 3, "Max retries(not including the first RPC)");
@@ -36,6 +40,25 @@ namespace brpc {
DECLARE_bool(http_verbose);
}
+class PartDataReader: public brpc::ProgressiveReader {
+public:
+ explicit PartDataReader(bthread::CountdownEvent* done): _done(done){}
+
+ butil::Status OnReadOnePart(const void* data, size_t length) {
+ const std::string part(static_cast<const char*>(data), length);
+ LOG(INFO) << "data: " << part << " size: " << length;
+ return butil::Status::OK();
+ }
+
+ void OnEndOfMessage(const butil::Status& status) {
+ LOG(INFO) << "progressive read data final status : " << status;
+ _done->signal();
+ delete this;
+ }
+private:
+ bthread::CountdownEvent* _done;
+};
+
int main(int argc, char* argv[]) {
// Parse gflags. We recommend you to use gflags as well.
GFLAGS_NAMESPACE::ParseCommandLineFlags(&argc, &argv, true);
@@ -71,6 +94,11 @@ int main(int argc, char* argv[]) {
cntl.request_attachment().append(FLAGS_d);
}
+ if (FLAGS_progressive) {
+
cntl.set_progressive_read_timeout_ms(FLAGS_progressive_read_timeout_ms);
+ cntl.response_will_be_read_progressively();
+ }
+
// Because `done'(last parameter) is NULL, this function waits until
// the response comes back or error occurs(including timedout).
channel.CallMethod(NULL, &cntl, NULL, NULL, NULL);
@@ -78,6 +106,13 @@ int main(int argc, char* argv[]) {
std::cerr << cntl.ErrorText() << std::endl;
return -1;
}
+
+ if (FLAGS_progressive) {
+ bthread::CountdownEvent done(1);
+ cntl.ReadProgressiveAttachmentBy(new PartDataReader(&done));
+ done.wait();
+ LOG(INFO) << "wait client progressive read done safely";
+ }
// If -http_verbose is on, brpc already prints the response to stderr.
if (!brpc::FLAGS_http_verbose) {
std::cout << cntl.response_attachment() << std::endl;
diff --git a/example/http_c++/http_server.cpp b/example/http_c++/http_server.cpp
index 05c9a0ee..4c3c8722 100644
--- a/example/http_c++/http_server.cpp
+++ b/example/http_c++/http_server.cpp
@@ -31,6 +31,7 @@ DEFINE_int32(idle_timeout_s, -1, "Connection will be closed
if there is no "
DEFINE_string(certificate, "cert.pem", "Certificate file path to enable SSL");
DEFINE_string(private_key, "key.pem", "Private key file path to enable SSL");
DEFINE_string(ciphers, "", "Cipher suite used for SSL connections");
+DEFINE_bool(enable_progressive_timeout, false, "whether or not trigger
progressive write attachment data timeout");
namespace example {
@@ -104,6 +105,9 @@ public:
// sleep a while to send another part.
bthread_usleep(10000);
+ if (FLAGS_enable_progressive_timeout && i > 50) {
+ bthread_usleep(100000000UL);
+ }
}
return NULL;
}
@@ -194,6 +198,9 @@ public:
// sleep a while to send another part.
bthread_usleep(10000 * 10);
+ if (FLAGS_enable_progressive_timeout && i > 50) {
+ bthread_usleep(100000000UL);
+ }
}
return NULL;
}
diff --git a/src/brpc/controller.cpp b/src/brpc/controller.cpp
index 8a8410be..66a9a3a5 100644
--- a/src/brpc/controller.cpp
+++ b/src/brpc/controller.cpp
@@ -73,6 +73,7 @@ BAIDU_REGISTER_ERRNO(brpc::EEOF, "Got EOF");
BAIDU_REGISTER_ERRNO(brpc::EUNUSED, "The socket was not needed");
BAIDU_REGISTER_ERRNO(brpc::ESSL, "SSL related operation failed");
BAIDU_REGISTER_ERRNO(brpc::EH2RUNOUTSTREAMS, "The H2 socket was run out of
streams");
+BAIDU_REGISTER_ERRNO(brpc::EPROGREADTIMEOUT, "Progressive read timed out");
BAIDU_REGISTER_ERRNO(brpc::EINTERNAL, "General internal error");
BAIDU_REGISTER_ERRNO(brpc::ERESPONSE, "Bad response");
@@ -94,8 +95,9 @@ namespace brpc {
DEFINE_bool(graceful_quit_on_sigterm, false,
"Register SIGTERM handle func to quit graceful");
DEFINE_bool(graceful_quit_on_sighup, false,
- "Register SIGHUP handle func to quit graceful");
-
+ "Register SIGHUP handle func to quit graceful");
+DEFINE_bool(log_idle_progressive_read_close, false,
+ "Print log when an idle progressive read is closed");
const IdlNames idl_single_req_single_res = { "req", "res" };
const IdlNames idl_single_req_multi_res = { "req", "" };
const IdlNames idl_multi_req_single_res = { "", "res" };
@@ -174,6 +176,226 @@ public:
void OnEndOfMessage(const butil::Status&) {}
};
+struct ProgressiveReadTimeoutTask;
+
+struct ProgressiveReadTimeoutState {
+ ProgressiveReadTimeoutState(SocketId id, int32_t timeout_ms)
+ : socket_id(id)
+ , read_timeout_ms(timeout_ms)
+ , deadline_us(butil::cpuwide_time_us() + timeout_ms * 1000L)
+ , timer_id(0)
+ , timer_task(NULL)
+ , user_callback_running(false)
+ , reader_failed(false)
+ , timeout_triggered(false)
+ , end_delivered(false) {}
+
+ butil::Mutex mutex;
+ const SocketId socket_id;
+ const int32_t read_timeout_ms;
+ int64_t deadline_us;
+ bthread_timer_t timer_id;
+ ProgressiveReadTimeoutTask* timer_task;
+ bool user_callback_running;
+ bool reader_failed;
+ bool timeout_triggered;
+ bool end_delivered;
+ butil::Status timer_error;
+};
+
+struct ProgressiveReadTimeoutTask {
+ explicit ProgressiveReadTimeoutTask(
+ const std::shared_ptr<ProgressiveReadTimeoutState>& state_in)
+ : state(state_in) {}
+
+ std::shared_ptr<ProgressiveReadTimeoutState> state;
+};
+
+class ProgressiveTimeoutReader : public ProgressiveReader {
+public:
+ ProgressiveTimeoutReader(SocketId id, int32_t read_timeout_ms,
+ ProgressiveReader* reader)
+ : _reader(reader)
+ , _state(new ProgressiveReadTimeoutState(id, read_timeout_ms)) {}
+
+ int Start() {
+ std::unique_lock<butil::Mutex> mu(_state->mutex);
+ return AddWatchdogLocked(_state, _state->read_timeout_ms * 1000L);
+ }
+
+ butil::Status OnReadOnePart(const void* data, size_t length) override {
+ {
+ std::unique_lock<butil::Mutex> mu(_state->mutex);
+ if (_state->timeout_triggered) {
+ return MakeTimeoutStatus(_state->read_timeout_ms);
+ }
+ if (!_state->timer_error.ok()) {
+ return _state->timer_error;
+ }
+ _state->user_callback_running = true;
+ }
+
+ butil::Status status = _reader->OnReadOnePart(data, length);
+ {
+ std::unique_lock<butil::Mutex> mu(_state->mutex);
+ _state->user_callback_running = false;
+ if (_state->timeout_triggered) {
+ status = MakeTimeoutStatus(_state->read_timeout_ms);
+ } else if (!_state->timer_error.ok()) {
+ status = _state->timer_error;
+ } else if (status.ok() && !_state->end_delivered) {
+ _state->deadline_us = butil::cpuwide_time_us() +
+ _state->read_timeout_ms * 1000L;
+ } else if (!status.ok()) {
+ _state->reader_failed = true;
+ }
+ }
+ return status;
+ }
+
+ void OnEndOfMessage(const butil::Status& status) override {
+ bthread_timer_t timer_id = 0;
+ ProgressiveReadTimeoutTask* timer_task = NULL;
+ butil::Status final_status = status;
+ ProgressiveReader* reader = NULL;
+ {
+ std::unique_lock<butil::Mutex> mu(_state->mutex);
+ if (_state->end_delivered) {
+ LOG(ERROR) << "ProgressiveReader::OnEndOfMessage was called
more than once";
+ return;
+ }
+ _state->end_delivered = true;
+ timer_id = _state->timer_id;
+ timer_task = _state->timer_task;
+ _state->timer_id = 0;
+ _state->timer_task = NULL;
+ if (_state->timeout_triggered) {
+ final_status = MakeTimeoutStatus(_state->read_timeout_ms);
+ } else if (!_state->timer_error.ok()) {
+ final_status = _state->timer_error;
+ }
+ reader = _reader;
+ _reader = NULL;
+ }
+
+ CancelWatchdog(timer_id, timer_task);
+ reader->OnEndOfMessage(final_status);
+ delete this;
+ }
+
+private:
+ ~ProgressiveTimeoutReader() override {}
+
+ static butil::Status MakeTimeoutStatus(int32_t timeout_ms) {
+ return butil::Status(
+ EPROGREADTIMEOUT,
+ "Progressive read timed out after %d ms", timeout_ms);
+ }
+
+ static butil::Status MakeTimerErrorStatus(int error_code) {
+ return butil::Status(
+ error_code, "Fail to add progressive read timeout timer: %s",
+ berror(error_code));
+ }
+
+ static void CancelWatchdog(
+ bthread_timer_t timer_id, ProgressiveReadTimeoutTask* timer_task) {
+ if (timer_id == 0) {
+ return;
+ }
+ const int rc = bthread_timer_del(timer_id);
+ if (rc == 0) {
+ delete timer_task;
+ } else if (rc == 1 || rc == EINVAL) {
+ // The callback owns timer_task once it starts running. EINVAL
means
+ // that the callback has already finished and released the task.
+ } else {
+ LOG(ERROR) << "Unexpected bthread_timer_del error=" << rc;
+ }
+ }
+
+ static int AddWatchdogLocked(
+ const std::shared_ptr<ProgressiveReadTimeoutState>& state,
+ int64_t delay_us) {
+ if (state->end_delivered || state->reader_failed) {
+ return ECANCELED;
+ }
+ if (delay_us <= 0) {
+ delay_us = 1;
+ }
+ ProgressiveReadTimeoutTask* task =
+ new (std::nothrow) ProgressiveReadTimeoutTask(state);
+ if (task == NULL) {
+ return ENOMEM;
+ }
+ bthread_timer_t timer_id = 0;
+ const int rc = bthread_timer_add(
+ &timer_id, butil::microseconds_from_now(delay_us),
+ HandleIdleProgressiveReader, task);
+ if (rc != 0) {
+ delete task;
+ return rc;
+ }
+ state->timer_id = timer_id;
+ state->timer_task = task;
+ return 0;
+ }
+
+ static void HandleIdleProgressiveReader(void* arg) {
+ std::unique_ptr<ProgressiveReadTimeoutTask> task(
+ static_cast<ProgressiveReadTimeoutTask*>(arg));
+ const std::shared_ptr<ProgressiveReadTimeoutState> state = task->state;
+ bool fail_socket = false;
+ int error_code = 0;
+ std::string error_text;
+ {
+ std::unique_lock<butil::Mutex> mu(state->mutex);
+ if (state->timer_task == task.get()) {
+ state->timer_id = 0;
+ state->timer_task = NULL;
+ }
+ if (state->end_delivered || state->reader_failed) {
+ return;
+ }
+
+ const int64_t now_us = butil::cpuwide_time_us();
+ if (state->user_callback_running || now_us < state->deadline_us) {
+ const int64_t delay_us = state->user_callback_running
+ ? state->read_timeout_ms * 1000L
+ : state->deadline_us - now_us;
+ const int rc = AddWatchdogLocked(state, delay_us);
+ if (rc != 0) {
+ state->timer_error = MakeTimerErrorStatus(rc);
+ fail_socket = true;
+ error_code = rc;
+ error_text = state->timer_error.error_str();
+ }
+ } else {
+ state->timeout_triggered = true;
+ fail_socket = true;
+ error_code = EPROGREADTIMEOUT;
+ error_text =
MakeTimeoutStatus(state->read_timeout_ms).error_str();
+ }
+ }
+
+ if (!fail_socket) {
+ return;
+ }
+ SocketUniquePtr socket;
+ if (Socket::Address(state->socket_id, &socket) != 0) {
+ LOG(ERROR) << "Fail to address socket_id=" << state->socket_id
+ << " after progressive read timeout";
+ } else {
+ LOG_IF(INFO, FLAGS_log_idle_progressive_read_close)
+ << error_text << ", socket_id=" << state->socket_id;
+ socket->SetFailed(error_code, "%s", error_text.c_str());
+ }
+ }
+
+ ProgressiveReader* _reader;
+ const std::shared_ptr<ProgressiveReadTimeoutState> _state;
+};
+
static IgnoreAllRead* s_ignore_all_read = NULL;
static pthread_once_t s_ignore_all_read_once = PTHREAD_ONCE_INIT;
static void CreateIgnoreAllRead() { s_ignore_all_read = new IgnoreAllRead; }
@@ -261,6 +483,7 @@ void Controller::ResetPods() {
_backup_request_ms = UNSET_MAGIC_NUM;
_backup_request_policy = NULL;
_connect_timeout_ms = UNSET_MAGIC_NUM;
+ _progressive_read_timeout_ms = UNSET_MAGIC_NUM;
_real_timeout_ms = UNSET_MAGIC_NUM;
_deadline_us = -1;
_timeout_id = 0;
@@ -336,6 +559,11 @@ void Controller::Call::Reset() {
stream_user_data = NULL;
}
+void Controller::set_progressive_read_timeout_ms(
+ int32_t progressive_read_timeout_ms) {
+ _progressive_read_timeout_ms = progressive_read_timeout_ms;
+}
+
void Controller::set_timeout_ms(int64_t timeout_ms) {
if (timeout_ms <= 0x7fffffff) {
_timeout_ms = timeout_ms;
@@ -1613,6 +1841,33 @@ void
Controller::ReadProgressiveAttachmentBy(ProgressiveReader* r) {
__FUNCTION__));
}
add_flag(FLAGS_PROGRESSIVE_READER);
+ if (progressive_read_timeout_ms() > 0) {
+ const SocketId socket_id = _rpa->GetSocketId();
+ if (socket_id == INVALID_SOCKET_ID) {
+ pthread_once(&s_ignore_all_read_once, CreateIgnoreAllRead);
+ _rpa->ReadProgressiveAttachmentBy(s_ignore_all_read);
+ return r->OnEndOfMessage(butil::Status(
+ ENOTSUP,
+ "Progressive read timeout is only supported for HTTP/1.x"));
+ }
+ ProgressiveTimeoutReader* reader = new (std::nothrow)
+ ProgressiveTimeoutReader(
+ socket_id, _progressive_read_timeout_ms, r);
+ if (reader == NULL) {
+ pthread_once(&s_ignore_all_read_once, CreateIgnoreAllRead);
+ _rpa->ReadProgressiveAttachmentBy(s_ignore_all_read);
+ return r->OnEndOfMessage(
+ butil::Status(ENOMEM, "Fail to create progressive timeout
reader"));
+ }
+ const int rc = reader->Start();
+ if (rc != 0) {
+ pthread_once(&s_ignore_all_read_once, CreateIgnoreAllRead);
+ _rpa->ReadProgressiveAttachmentBy(s_ignore_all_read);
+ return reader->OnEndOfMessage(butil::Status(
+ rc, "Fail to add progressive read timeout timer: %s",
berror(rc)));
+ }
+ return _rpa->ReadProgressiveAttachmentBy(reader);
+ }
return _rpa->ReadProgressiveAttachmentBy(r);
}
diff --git a/src/brpc/controller.h b/src/brpc/controller.h
index 564c0875..41d42ae7 100644
--- a/src/brpc/controller.h
+++ b/src/brpc/controller.h
@@ -199,6 +199,9 @@ public:
// Set/get timeout in milliseconds for the RPC call. Use
// ChannelOptions.timeout_ms on unset.
+ void set_progressive_read_timeout_ms(int32_t progressive_read_timeout_ms);
+ int32_t progressive_read_timeout_ms() const { return
_progressive_read_timeout_ms; }
+
void set_timeout_ms(int64_t timeout_ms);
int64_t timeout_ms() const { return _timeout_ms; }
@@ -361,7 +364,9 @@ public:
// Make the RPC end when the HTTP response has complete headers and let
// user read the remaining body by using ReadProgressiveAttachmentBy().
- void response_will_be_read_progressively() {
add_flag(FLAGS_READ_PROGRESSIVELY); }
+ void response_will_be_read_progressively() {
+ add_flag(FLAGS_READ_PROGRESSIVELY);
+ }
// Make the RPC end when the HTTP request has complete headers and let
// user read the remaining body by using ReadProgressiveAttachmentBy().
void request_will_be_read_progressively() {
add_flag(FLAGS_READ_PROGRESSIVELY); }
@@ -911,6 +916,7 @@ private:
int32_t _timeout_ms;
int32_t _connect_timeout_ms;
int32_t _backup_request_ms;
+ int32_t _progressive_read_timeout_ms;
// Priority: `_backup_request_policy' > `_backup_request_ms'.
BackupRequestPolicy* _backup_request_policy;
// If this rpc call has retry/backup request,this var save the real
timeout for current call
diff --git a/src/brpc/errno.proto b/src/brpc/errno.proto
index 26ffadc2..166d82dc 100644
--- a/src/brpc/errno.proto
+++ b/src/brpc/errno.proto
@@ -41,6 +41,7 @@ enum Errno {
ESSL = 1016; // SSL related error
EH2RUNOUTSTREAMS = 1017; // The H2 socket was run out of streams
EREJECT = 1018; // The Request is rejected
+ EPROGREADTIMEOUT = 1019; // The Progressive read timeout
// Errno caused by server
EINTERNAL = 2001; // Internal Server Error
diff --git a/src/brpc/policy/http_rpc_protocol.cpp
b/src/brpc/policy/http_rpc_protocol.cpp
index 8cbe0698..3fb94088 100644
--- a/src/brpc/policy/http_rpc_protocol.cpp
+++ b/src/brpc/policy/http_rpc_protocol.cpp
@@ -1201,6 +1201,7 @@ ParseResult ParseHttpMessage(butil::IOBuf *source, Socket
*socket,
LOG(FATAL) << "Fail to new HttpContext";
return MakeParseError(PARSE_ERROR_NO_RESOURCE);
}
+ http_imsg->SetSocketId(socket->id());
// Parsing http is costly, parsing an incomplete http message from the
// beginning repeatedly should be avoided, otherwise the cost may reach
// O(n^2) in the worst case. Save incomplete http messages in sockets
diff --git a/src/brpc/policy/http_rpc_protocol.h
b/src/brpc/policy/http_rpc_protocol.h
index bc8bd065..cd41798e 100644
--- a/src/brpc/policy/http_rpc_protocol.h
+++ b/src/brpc/policy/http_rpc_protocol.h
@@ -87,11 +87,20 @@ class HttpContext : public ReadableProgressiveAttachment
, public InputMessageBase
, public HttpMessage {
public:
+ SocketId GetSocketId() override {
+ return _socket_id;
+ }
+
+ void SetSocketId(SocketId id) {
+ _socket_id = id;
+ }
+
explicit HttpContext(bool read_body_progressively,
HttpMethod request_method = HTTP_METHOD_GET)
: InputMessageBase()
, HttpMessage(read_body_progressively, request_method)
- , _is_stage2(false) {
+ , _is_stage2(false)
+ , _socket_id(INVALID_SOCKET_ID) {
// add one ref for Destroy
butil::intrusive_ptr<HttpContext>(this).detach();
}
@@ -122,6 +131,7 @@ public:
private:
bool _is_stage2;
+ SocketId _socket_id;
};
// Implement functions required in protocol.h
diff --git a/src/brpc/progressive_reader.h b/src/brpc/progressive_reader.h
index 6f54ae68..c84be8b7 100644
--- a/src/brpc/progressive_reader.h
+++ b/src/brpc/progressive_reader.h
@@ -20,6 +20,7 @@
#define BRPC_PROGRESSIVE_READER_H
#include "brpc/shared_object.h"
+#include "brpc/socket_id.h"
namespace brpc {
@@ -84,6 +85,7 @@ public:
// Any error occurred should destroy the reader by calling r->Destroy().
// r->Destroy() should be guaranteed to be called once and only once.
virtual void ReadProgressiveAttachmentBy(ProgressiveReader* r) = 0;
+ virtual SocketId GetSocketId() = 0;
};
} // namespace brpc
diff --git a/test/brpc_http_rpc_protocol_unittest.cpp
b/test/brpc_http_rpc_protocol_unittest.cpp
index 87837abf..6735a171 100644
--- a/test/brpc_http_rpc_protocol_unittest.cpp
+++ b/test/brpc_http_rpc_protocol_unittest.cpp
@@ -19,6 +19,7 @@
// Date: Sun Jul 13 15:04:18 CST 2014
+#include <atomic>
#include <cstddef>
#include <string>
#include <sys/ioctl.h>
@@ -736,9 +737,13 @@ static void CopyPAPrefixedWithSeqNo(char* buf, uint64_t
seq_no) {
class DownloadServiceImpl : public ::test::DownloadService {
public:
DownloadServiceImpl(DonePlace done_place = DONE_BEFORE_CREATE_PA,
- size_t num_repeat = 1)
+ size_t num_repeat = 1,
+ int write_interval_us = 0,
+ int initial_write_delay_us = 0)
: _done_place(done_place)
, _nrep(num_repeat)
+ , _write_interval_us(write_interval_us)
+ , _initial_write_delay_us(initial_write_delay_us)
, _nwritten(0)
, _ever_full(false)
, _last_errno(0) {}
@@ -762,6 +767,9 @@ public:
if (_done_place == DONE_BEFORE_CREATE_PA) {
done_guard.reset(NULL);
}
+ if (_initial_write_delay_us > 0) {
+ bthread_usleep(_initial_write_delay_us);
+ }
ASSERT_GT(PA_DATA_LEN, 8u); // long enough to hold a 64-bit decimal.
char buf[PA_DATA_LEN];
for (size_t c = 0; c < _nrep;) {
@@ -778,6 +786,9 @@ public:
}
} else {
_nwritten += PA_DATA_LEN;
+ if (_write_interval_us > 0) {
+ bthread_usleep(_write_interval_us);
+ }
}
++c;
}
@@ -840,6 +851,8 @@ public:
private:
DonePlace _done_place;
size_t _nrep;
+ int _write_interval_us;
+ int _initial_write_delay_us;
size_t _nwritten;
bool _ever_full;
int _last_errno;
@@ -941,6 +954,47 @@ private:
butil::Status _destroying_st;
};
+class TimeoutReadBody : public brpc::ProgressiveReader,
+ public brpc::SharedObject {
+public:
+ explicit TimeoutReadBody(int read_delay_us = 0, int read_error = 0)
+ : _read_delay_us(read_delay_us)
+ , _read_error(read_error)
+ , _nread(0)
+ , _nend(0)
+ , _end_error(0) {
+ butil::intrusive_ptr<TimeoutReadBody>(this).detach();
+ }
+
+ butil::Status OnReadOnePart(const void*, size_t length) override {
+ if (_read_delay_us > 0) {
+ bthread_usleep(_read_delay_us);
+ }
+ _nread.fetch_add(length);
+ if (_read_error != 0) {
+ return butil::Status(_read_error, "intended progressive read
failure");
+ }
+ return butil::Status::OK();
+ }
+
+ void OnEndOfMessage(const butil::Status& status) override {
+ _end_error.store(status.error_code());
+ _nend.fetch_add(1);
+ butil::intrusive_ptr<TimeoutReadBody>(this, false);
+ }
+
+ size_t read_bytes() const { return _nread.load(); }
+ int end_count() const { return _nend.load(); }
+ int end_error() const { return _end_error.load(); }
+
+private:
+ const int _read_delay_us;
+ const int _read_error;
+ std::atomic<size_t> _nread;
+ std::atomic<int> _nend;
+ std::atomic<int> _end_error;
+};
+
#ifdef BUTIL_USE_ASAN
static const int GENERAL_DELAY_US = 1000000; // 1s
#else
@@ -1034,6 +1088,187 @@ TEST_F(HttpTest, read_short_body_progressively) {
}
}
+TEST_F(HttpTest, progressive_read_timeout_keeps_active_reader_alive) {
+ const int port = 8923;
+ brpc::Server server;
+ DownloadServiceImpl svc(DONE_BEFORE_CREATE_PA, 8, 100000);
+ ASSERT_EQ(0, server.AddService(&svc, brpc::SERVER_DOESNT_OWN_SERVICE));
+ ASSERT_EQ(0, server.Start(port, NULL));
+
+ brpc::Channel channel;
+ brpc::ChannelOptions options;
+ options.protocol = brpc::PROTOCOL_HTTP;
+ ASSERT_EQ(0, channel.Init(butil::EndPoint(butil::my_ip(), port),
&options));
+
+ brpc::Controller cntl;
+ cntl.response_will_be_read_progressively();
+ cntl.set_progressive_read_timeout_ms(500);
+ cntl.http_request().uri() = "/DownloadService/Download";
+ channel.CallMethod(NULL, &cntl, NULL, NULL, NULL);
+ ASSERT_FALSE(cntl.Failed()) << cntl.ErrorText();
+
+ butil::intrusive_ptr<TimeoutReadBody> reader(new TimeoutReadBody);
+ cntl.ReadProgressiveAttachmentBy(reader.get());
+ for (int i = 0; i < 200 && reader->end_count() == 0; ++i) {
+ bthread_usleep(10000);
+ }
+ ASSERT_EQ(1, reader->end_count());
+ EXPECT_EQ(0, reader->end_error());
+ EXPECT_EQ(8 * PA_DATA_LEN, reader->read_bytes());
+}
+
+TEST_F(HttpTest, progressive_read_timeout_closes_idle_http1_reader_once) {
+ const int port = 8923;
+ brpc::Server server;
+ DownloadServiceImpl svc(DONE_BEFORE_CREATE_PA, 2, 300000);
+ ASSERT_EQ(0, server.AddService(&svc, brpc::SERVER_DOESNT_OWN_SERVICE));
+ ASSERT_EQ(0, server.Start(port, NULL));
+
+ butil::intrusive_ptr<TimeoutReadBody> reader(new TimeoutReadBody);
+ {
+ brpc::Channel channel;
+ brpc::ChannelOptions options;
+ options.protocol = brpc::PROTOCOL_HTTP;
+ ASSERT_EQ(0, channel.Init(butil::EndPoint(butil::my_ip(), port),
&options));
+ {
+ brpc::Controller cntl;
+ cntl.response_will_be_read_progressively();
+ cntl.set_progressive_read_timeout_ms(50);
+ cntl.http_request().uri() = "/DownloadService/Download";
+ channel.CallMethod(NULL, &cntl, NULL, NULL, NULL);
+ ASSERT_FALSE(cntl.Failed()) << cntl.ErrorText();
+ cntl.ReadProgressiveAttachmentBy(reader.get());
+ bthread_usleep(400000);
+ ASSERT_NE(0, svc.last_errno());
+ EXPECT_EQ(0, reader->end_count());
+ }
+ }
+ for (int i = 0; i < 100 && reader->end_count() == 0; ++i) {
+ bthread_usleep(10000);
+ }
+ ASSERT_EQ(1, reader->end_count());
+ EXPECT_EQ(brpc::EPROGREADTIMEOUT, reader->end_error());
+ bthread_usleep(400000);
+ EXPECT_EQ(1, reader->end_count());
+}
+
+TEST_F(HttpTest, progressive_read_timeout_before_first_body_part) {
+ const int port = 8923;
+ brpc::Server server;
+ DownloadServiceImpl svc(DONE_BEFORE_CREATE_PA, 1, 0, 300000);
+ ASSERT_EQ(0, server.AddService(&svc, brpc::SERVER_DOESNT_OWN_SERVICE));
+ ASSERT_EQ(0, server.Start(port, NULL));
+
+ butil::intrusive_ptr<TimeoutReadBody> reader(new TimeoutReadBody);
+ {
+ brpc::Channel channel;
+ brpc::ChannelOptions options;
+ options.protocol = brpc::PROTOCOL_HTTP;
+ ASSERT_EQ(0, channel.Init(butil::EndPoint(butil::my_ip(), port),
&options));
+ {
+ brpc::Controller cntl;
+ cntl.response_will_be_read_progressively();
+ cntl.set_progressive_read_timeout_ms(50);
+ cntl.http_request().uri() = "/DownloadService/Download";
+ channel.CallMethod(NULL, &cntl, NULL, NULL, NULL);
+ ASSERT_FALSE(cntl.Failed()) << cntl.ErrorText();
+ cntl.ReadProgressiveAttachmentBy(reader.get());
+ bthread_usleep(400000);
+ ASSERT_NE(0, svc.last_errno());
+ EXPECT_EQ(size_t(0), reader->read_bytes());
+ EXPECT_EQ(0, reader->end_count());
+ }
+ }
+ for (int i = 0; i < 100 && reader->end_count() == 0; ++i) {
+ bthread_usleep(10000);
+ }
+ ASSERT_EQ(1, reader->end_count());
+ EXPECT_EQ(brpc::EPROGREADTIMEOUT, reader->end_error());
+}
+
+TEST_F(HttpTest, progressive_read_timeout_ignores_slow_user_callback) {
+ const int port = 8923;
+ brpc::Server server;
+ DownloadServiceImpl svc(DONE_BEFORE_CREATE_PA, 3, 50000);
+ ASSERT_EQ(0, server.AddService(&svc, brpc::SERVER_DOESNT_OWN_SERVICE));
+ ASSERT_EQ(0, server.Start(port, NULL));
+
+ brpc::Channel channel;
+ brpc::ChannelOptions options;
+ options.protocol = brpc::PROTOCOL_HTTP;
+ ASSERT_EQ(0, channel.Init(butil::EndPoint(butil::my_ip(), port),
&options));
+
+ brpc::Controller cntl;
+ cntl.response_will_be_read_progressively();
+ cntl.set_progressive_read_timeout_ms(50);
+ cntl.http_request().uri() = "/DownloadService/Download";
+ channel.CallMethod(NULL, &cntl, NULL, NULL, NULL);
+ ASSERT_FALSE(cntl.Failed()) << cntl.ErrorText();
+
+ butil::intrusive_ptr<TimeoutReadBody> reader(
+ new TimeoutReadBody(200000));
+ cntl.ReadProgressiveAttachmentBy(reader.get());
+ for (int i = 0; i < 100 && reader->end_count() == 0; ++i) {
+ bthread_usleep(10000);
+ }
+ ASSERT_EQ(1, reader->end_count());
+ EXPECT_EQ(0, reader->end_error());
+ EXPECT_EQ(3 * PA_DATA_LEN, reader->read_bytes());
+}
+
+TEST_F(HttpTest, progressive_read_timeout_preserves_reader_error) {
+ const int port = 8923;
+ brpc::Server server;
+ DownloadServiceImpl svc(DONE_BEFORE_CREATE_PA, 10);
+ ASSERT_EQ(0, server.AddService(&svc, brpc::SERVER_DOESNT_OWN_SERVICE));
+ ASSERT_EQ(0, server.Start(port, NULL));
+
+ brpc::Channel channel;
+ brpc::ChannelOptions options;
+ options.protocol = brpc::PROTOCOL_HTTP;
+ ASSERT_EQ(0, channel.Init(butil::EndPoint(butil::my_ip(), port),
&options));
+
+ brpc::Controller cntl;
+ cntl.response_will_be_read_progressively();
+ cntl.set_progressive_read_timeout_ms(1000);
+ cntl.http_request().uri() = "/DownloadService/Download";
+ channel.CallMethod(NULL, &cntl, NULL, NULL, NULL);
+ ASSERT_FALSE(cntl.Failed()) << cntl.ErrorText();
+
+ butil::intrusive_ptr<TimeoutReadBody> reader(
+ new TimeoutReadBody(0, EIO));
+ cntl.ReadProgressiveAttachmentBy(reader.get());
+ ASSERT_EQ(1, reader->end_count());
+ EXPECT_EQ(EIO, reader->end_error());
+}
+
+TEST_F(HttpTest, progressive_read_timeout_rejects_http2) {
+ const int port = 8923;
+ brpc::Server server;
+ ASSERT_EQ(0, server.AddService(&_svc, brpc::SERVER_DOESNT_OWN_SERVICE));
+ ASSERT_EQ(0, server.Start(port, NULL));
+
+ brpc::Channel channel;
+ brpc::ChannelOptions options;
+ options.protocol = brpc::PROTOCOL_H2;
+ ASSERT_EQ(0, channel.Init(butil::EndPoint(butil::my_ip(), port),
&options));
+
+ brpc::Controller cntl;
+ cntl.response_will_be_read_progressively();
+ cntl.set_progressive_read_timeout_ms(1000);
+ cntl.http_request().uri() = "/EchoService/Echo";
+ test::EchoRequest req;
+ req.set_message(EXP_REQUEST);
+ channel.CallMethod(NULL, &cntl, &req, NULL, NULL);
+ ASSERT_FALSE(cntl.Failed()) << cntl.ErrorText();
+
+ butil::intrusive_ptr<TimeoutReadBody> reader(new TimeoutReadBody);
+ cntl.ReadProgressiveAttachmentBy(reader.get());
+ ASSERT_EQ(1, reader->end_count());
+ EXPECT_EQ(ENOTSUP, reader->end_error());
+ EXPECT_EQ(size_t(0), reader->read_bytes());
+}
+
TEST_F(HttpTest, read_progressively_after_cntl_destroys) {
DownloadServiceImpl svc(DONE_BEFORE_CREATE_PA,
std::numeric_limits<size_t>::max());
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
