Merge pull request #849 from zyearn/ignore_flowcontrol_in_first_req

ignore flow control in h2 when sending first request

src/brpc/policy/http2_rpc_protocol.cpp

@@ -326,14 +326,20 @@ inline H2Context::FrameHandler FindFrameHandler(H2FrameType type) {
 
 H2Context::H2Context(Socket* socket, const Server* server)
     : _socket(socket)
-    , _remote_window_left(H2Settings::DEFAULT_INITIAL_WINDOW_SIZE)
+    // Maximize the window size to make sending big request possible before
+    // receving the remote settings.
+    , _remote_window_left(H2Settings::MAX_WINDOW_SIZE)
     , _conn_state(H2_CONNECTION_UNINITIALIZED)
     , _last_received_stream_id(-1)
     , _last_sent_stream_id(1)
     , _goaway_stream_id(-1)
+    , _remote_settings_received(false)
     , _deferred_window_update(0) {
     // Stop printing the field which is useless for remote settings.
     _remote_settings.connection_window_size = 0;
+    // Maximize the window size to make sending big request possible before
+    // receving the remote settings.
+    _remote_settings.stream_window_size = H2Settings::MAX_WINDOW_SIZE;
     if (server) {
         _unack_local_settings = server->options().h2_settings;
     } else {
@@ -860,9 +866,28 @@ H2ParseResult H2Context::OnSettings(
         return MakeH2Message(NULL);
     }
     const int64_t old_stream_window_size = _remote_settings.stream_window_size;
-    if (!ParseH2Settings(&_remote_settings, it, frame_head.payload_size)) {
-        LOG(ERROR) << "Fail to parse from SETTINGS";
-        return MakeH2Error(H2_PROTOCOL_ERROR);
+    if (!_remote_settings_received) {
+        // To solve the problem that sender can't send large request before receving
+        // remote setting, the initial window size of stream/connection is set to
+        // MAX_WINDOW_SIZE(see constructor of H2Context).
+        // As a result, in the view of remote side, window size is 65535 by default so
+        // it may not send its stream size to sender, making stream size still be
+        // MAX_WINDOW_SIZE. In this case we need to revert this value to default.
+        H2Settings tmp_settings;
+        if (!ParseH2Settings(&tmp_settings, it, frame_head.payload_size)) {
+            LOG(ERROR) << "Fail to parse from SETTINGS";
+            return MakeH2Error(H2_PROTOCOL_ERROR);
+        }
+        _remote_settings = tmp_settings;
+        _remote_window_left.fetch_sub(
+                H2Settings::MAX_WINDOW_SIZE - H2Settings::DEFAULT_INITIAL_WINDOW_SIZE,
+                butil::memory_order_relaxed);
+        _remote_settings_received = true;
+    } else {
+        if (!ParseH2Settings(&_remote_settings, it, frame_head.payload_size)) {
+            LOG(ERROR) << "Fail to parse from SETTINGS";
+            return MakeH2Error(H2_PROTOCOL_ERROR);
+        }
     }
     const int64_t window_diff =
         static_cast<int64_t>(_remote_settings.stream_window_size)
@@ -1025,6 +1050,7 @@ void H2Context::Describe(std::ostream& os, const DescribeOptions& opt) const {
        << sep << "remote_conn_window_left="
        << _remote_window_left.load(butil::memory_order_relaxed)
        << sep << "remote_settings=" << _remote_settings
+       << sep << "remote_settings_received=" << _remote_settings_received
        << sep << "local_settings=" << _local_settings
        << sep << "hpacker={";
     IndentingOStream os2(os, 2);
@@ -1527,7 +1553,7 @@ H2UnsentRequest::AppendAndDestroySelf(butil::IOBuf* out, Socket* socket) {
     }
 
     _sctx->Init(ctx, id);
-    // flow control
+    // check flow control restriction
     if (!_cntl->request_attachment().empty()) {
         const int64_t data_size = _cntl->request_attachment().size();
         if (!_sctx->ConsumeWindowSize(data_size)) {

src/brpc/policy/http2_rpc_protocol.h

@@ -387,6 +387,7 @@ friend void InitFrameHandlers();
     uint32_t _last_sent_stream_id;
     int _goaway_stream_id;
     H2Settings _remote_settings;
+    bool _remote_settings_received;
     H2Settings _local_settings;
     H2Settings _unack_local_settings;
     HPacker _hpacker;

test/brpc_http_rpc_protocol_unittest.cpp

@@ -96,8 +96,6 @@ class MyEchoService : public ::test::EchoService {
         if (sleep_ms_str) {
             bthread_usleep(strtol(sleep_ms_str->data(), NULL, 10) * 1000);
         }
-
-        EXPECT_EQ(EXP_REQUEST, req->message());
         res->set_message(EXP_RESPONSE);
     }
 };
@@ -996,12 +994,24 @@ TEST_F(HttpTest, http2_sanity) {
     options.protocol = "h2";
     ASSERT_EQ(0, channel.Init(butil::EndPoint(butil::my_ip(), port), &options));
 
-    // 1) complete flow and
-    // 2) socket replacement when streamId runs out, the initial streamId is a special
-    // value set in ctor of H2Context
+    // Check that the first request with size larger than the default window can
+    // be sent out, when remote settings are not received.
+    brpc::Controller cntl;
+    test::EchoRequest big_req;
+    test::EchoResponse res;
+    std::string message(2 * 1024 * 1024 /* 2M */, 'x');
+    big_req.set_message(message);
+    cntl.http_request().set_method(brpc::HTTP_METHOD_POST);
+    cntl.http_request().uri() = "/EchoService/Echo";
+    channel.CallMethod(NULL, &cntl, &big_req, &res, NULL);
+    ASSERT_FALSE(cntl.Failed());
+    ASSERT_EQ(EXP_RESPONSE, res.message());
+
+    // socket replacement when streamId runs out, the initial streamId is a special
+    // value set in ctor of H2Context so that the number 15000 is enough to run out
+    // of stream.
     test::EchoRequest req;
     req.set_message(EXP_REQUEST);
-    test::EchoResponse res;
     for (int i = 0; i < 15000; ++i) {
         brpc::Controller cntl;
         cntl.http_request().set_content_type("application/json");
@@ -1113,6 +1123,14 @@ TEST_F(HttpTest, http2_window_used_up) {
     cntl.http_request().set_content_type("application/proto");
     brpc::policy::SerializeHttpRequest(&request_buf, &cntl, &req);
 
+    char settingsbuf[brpc::policy::FRAME_HEAD_SIZE + 36];
+    brpc::H2Settings h2_settings;
+    const size_t nb = brpc::policy::SerializeH2Settings(h2_settings, settingsbuf + brpc::policy::FRAME_HEAD_SIZE);
+    brpc::policy::SerializeFrameHead(settingsbuf, nb, brpc::policy::H2_FRAME_SETTINGS, 0, 0);
+    butil::IOBuf buf;
+    buf.append(settingsbuf, brpc::policy::FRAME_HEAD_SIZE + nb);
+    brpc::policy::ParseH2Message(&buf, _h2_client_sock.get(), false, NULL);
+
     int nsuc = brpc::H2Settings::DEFAULT_INITIAL_WINDOW_SIZE / cntl.request_attachment().size();
     for (int i = 0; i <= nsuc; i++) {
         brpc::policy::H2UnsentRequest* h2_req = brpc::policy::H2UnsentRequest::New(&cntl);