diff --git a/example/http_c++/http_client.cpp b/example/http_c++/http_client.cpp index 3a09186f84..23222dee9b 100644 --- a/example/http_c++/http_client.cpp +++ b/example/http_c++/http_client.cpp @@ -22,15 +22,11 @@ // - Access www.foo.com // ./http_client www.foo.com -#include #include #include #include -#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)"); @@ -40,25 +36,6 @@ 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(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); @@ -94,11 +71,6 @@ 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); @@ -106,13 +78,6 @@ 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 4c3c8722fd..05c9a0ee4c 100644 --- a/example/http_c++/http_server.cpp +++ b/example/http_c++/http_server.cpp @@ -31,7 +31,6 @@ 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 { @@ -105,9 +104,6 @@ class FileServiceImpl : public FileService { // sleep a while to send another part. bthread_usleep(10000); - if (FLAGS_enable_progressive_timeout && i > 50) { - bthread_usleep(100000000UL); - } } return NULL; } @@ -198,9 +194,6 @@ class HttpSSEServiceImpl : public HttpSSEService { // 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 66a9a3a504..8a8410beb9 100644 --- a/src/brpc/controller.cpp +++ b/src/brpc/controller.cpp @@ -73,7 +73,6 @@ 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"); @@ -95,9 +94,8 @@ 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"); -DEFINE_bool(log_idle_progressive_read_close, false, - "Print log when an idle progressive read is closed"); + "Register SIGHUP handle func to quit graceful"); + 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" }; @@ -176,226 +174,6 @@ class IgnoreAllRead : public ProgressiveReader { 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& state_in) - : state(state_in) {} - - std::shared_ptr 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 mu(_state->mutex); - return AddWatchdogLocked(_state, _state->read_timeout_ms * 1000L); - } - - butil::Status OnReadOnePart(const void* data, size_t length) override { - { - std::unique_lock 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 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 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& 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 task( - static_cast(arg)); - const std::shared_ptr state = task->state; - bool fail_socket = false; - int error_code = 0; - std::string error_text; - { - std::unique_lock 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 _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; } @@ -483,7 +261,6 @@ 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; @@ -559,11 +336,6 @@ 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; @@ -1841,33 +1613,6 @@ 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 41d42ae712..564c0875e1 100644 --- a/src/brpc/controller.h +++ b/src/brpc/controller.h @@ -199,9 +199,6 @@ friend void policy::ProcessThriftRequest(InputMessageBase*); // 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; } @@ -364,9 +361,7 @@ friend void policy::ProcessThriftRequest(InputMessageBase*); // 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); } @@ -916,7 +911,6 @@ friend void policy::ProcessThriftRequest(InputMessageBase*); 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 166d82dc4a..26ffadc201 100644 --- a/src/brpc/errno.proto +++ b/src/brpc/errno.proto @@ -41,7 +41,6 @@ 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 3fb9408850..8cbe06980f 100644 --- a/src/brpc/policy/http_rpc_protocol.cpp +++ b/src/brpc/policy/http_rpc_protocol.cpp @@ -1201,7 +1201,6 @@ 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 cd41798e9f..bc8bd06593 100644 --- a/src/brpc/policy/http_rpc_protocol.h +++ b/src/brpc/policy/http_rpc_protocol.h @@ -87,20 +87,11 @@ 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) - , _socket_id(INVALID_SOCKET_ID) { + , _is_stage2(false) { // add one ref for Destroy butil::intrusive_ptr(this).detach(); } @@ -131,7 +122,6 @@ class HttpContext : public ReadableProgressiveAttachment 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 c84be8b7e7..6f54ae68a7 100644 --- a/src/brpc/progressive_reader.h +++ b/src/brpc/progressive_reader.h @@ -20,7 +20,6 @@ #define BRPC_PROGRESSIVE_READER_H #include "brpc/shared_object.h" -#include "brpc/socket_id.h" namespace brpc { @@ -85,7 +84,6 @@ class ReadableProgressiveAttachment : public SharedObject { // 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 706913140f..97de699547 100644 --- a/test/brpc_http_rpc_protocol_unittest.cpp +++ b/test/brpc_http_rpc_protocol_unittest.cpp @@ -19,7 +19,6 @@ // Date: Sun Jul 13 15:04:18 CST 2014 -#include #include #include #include @@ -737,13 +736,9 @@ 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, - int write_interval_us = 0, - int initial_write_delay_us = 0) + size_t num_repeat = 1) : _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) {} @@ -767,9 +762,6 @@ class DownloadServiceImpl : public ::test::DownloadService { 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;) { @@ -786,9 +778,6 @@ class DownloadServiceImpl : public ::test::DownloadService { } } else { _nwritten += PA_DATA_LEN; - if (_write_interval_us > 0) { - bthread_usleep(_write_interval_us); - } } ++c; } @@ -851,8 +840,6 @@ class DownloadServiceImpl : public ::test::DownloadService { 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; @@ -954,47 +941,6 @@ class ReadBody : public brpc::ProgressiveReader, 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(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(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 _nread; - std::atomic _nend; - std::atomic _end_error; -}; - #ifdef BUTIL_USE_ASAN static const int GENERAL_DELAY_US = 1000000; // 1s #else @@ -1088,187 +1034,6 @@ 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 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 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 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 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 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 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::max());