Merge pull request #781 from zyearn/fix_http_response_after_h2goaway

fix wrong pointer in processing http response after h2goaway

src/brpc/policy/http2_rpc_protocol.cpp

@@ -967,8 +967,8 @@ H2ParseResult H2Context::OnGoAway(
                                   BTHREAD_ATTR_PTHREAD :
                                   BTHREAD_ATTR_NORMAL);
             tmp.keytable_pool = _socket->keytable_pool();
-            CHECK_EQ(0, bthread_start_background(
-                         &th, &tmp, ProcessHttpResponseWrapper, goaway_streams[i]));
+            CHECK_EQ(0, bthread_start_background(&th, &tmp, ProcessHttpResponseWrapper,
+                         static_cast<InputMessageBase*>(goaway_streams[i])));
         }
         return MakeH2Message(goaway_streams[0]);
     } else {

test/brpc_http_rpc_protocol_unittest.cpp

@@ -12,6 +12,7 @@
 #include "butil/time.h"
 #include "butil/macros.h"
 #include "butil/files/scoped_file.h"
+#include "butil/fd_guard.h"
 #include "brpc/socket.h"
 #include "brpc/acceptor.h"
 #include "brpc/server.h"
@@ -1324,7 +1325,7 @@ TEST_F(HttpTest, http2_header_after_data) {
     ASSERT_EQ(*user_defined2, "b");
 }
 
-TEST_F(HttpTest, http2_goaway) {
+TEST_F(HttpTest, http2_goaway_sanity) {
     brpc::Controller cntl;
     // Prepare request
     butil::IOBuf req_out;
@@ -1363,4 +1364,49 @@ TEST_F(HttpTest, http2_goaway) {
     ASSERT_TRUE(st.error_data().ends_with("the connection just issued GOAWAY"));
 }
 
+class AfterRecevingGoAway : public ::google::protobuf::Closure {
+public:
+    void Run() {
+        ASSERT_EQ(brpc::EHTTP, cntl.ErrorCode());
+        delete this;
+    }
+    brpc::Controller cntl;
+};
+
+TEST_F(HttpTest, http2_handle_goaway_streams) {
+    const butil::EndPoint ep(butil::IP_ANY, 5961);
+    butil::fd_guard listenfd(butil::tcp_listen(ep));
+    ASSERT_GT(listenfd, 0);
+
+    brpc::Channel channel;
+    brpc::ChannelOptions options;
+    options.protocol = brpc::PROTOCOL_H2;
+    ASSERT_EQ(0, channel.Init(ep, &options));
+
+    int req_size = 10;
+    std::vector<brpc::CallId> ids(req_size);
+    for (int i = 0; i < req_size; i++) {
+        AfterRecevingGoAway* done = new AfterRecevingGoAway;
+        brpc::Controller& cntl = done->cntl;
+        ids.push_back(cntl.call_id());
+        cntl.set_timeout_ms(-1);
+        cntl.http_request().uri() = "/it-doesnt-matter";
+        channel.CallMethod(NULL, &cntl, NULL, NULL, done);
+    }
+
+    int servfd = accept(listenfd, NULL, NULL);
+    ASSERT_GT(servfd, 0);
+    // Sleep for a while to make sure that server has received all data.
+    bthread_usleep(2000);
+    char goawaybuf[brpc::policy::FRAME_HEAD_SIZE + 8];
+    SerializeFrameHead(goawaybuf, 8, brpc::policy::H2_FRAME_GOAWAY, 0, 0);
+    SaveUint32(goawaybuf + brpc::policy::FRAME_HEAD_SIZE, 0);
+    SaveUint32(goawaybuf + brpc::policy::FRAME_HEAD_SIZE + 4, 0);
+    ASSERT_EQ(brpc::policy::FRAME_HEAD_SIZE + 8, ::write(servfd, goawaybuf, brpc::policy::FRAME_HEAD_SIZE + 8));
+
+    // After receving GOAWAY, the callbacks in client should be run correctly.
+    for (int i = 0; i < req_size; i++) {
+        brpc::Join(ids[i]);
+    }
+}
 } //namespace