Progressive timeout dev

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,13 +94,25 @@ 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);
     if (cntl.Failed()) {
         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;

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 @@ 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;
     }
@@ -194,6 +198,9 @@ 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;
     }

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 @@ 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<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;
@@ -1611,6 +1839,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);
 }