MPP: ExchangeReceiver supports async gRPC (#3715)

close #4298
pingcap · Mar 16, 2022 · 6e1356b · 6e1356b
1 parent 584b552
commit 6e1356b
Show file tree

Hide file tree

Showing 21 changed files with 899 additions and 10 deletions.
diff --git a/dbms/src/Common/TiFlashMetrics.h b/dbms/src/Common/TiFlashMetrics.h
@@ -179,6 +179,7 @@ namespace DB
         F(type_total_threads_of_thdpool, {"type", "thread_pool_total"}),                                                                  \
         F(type_max_threads_of_raw, {"type", "total_max"}),                                                                                \
         F(type_total_threads_of_raw, {"type", "total"}),                                                                                  \
+        F(type_threads_of_client_cq_pool, {"type", "rpc_client_cq_pool"}),                                                                \
         F(type_max_threads_of_establish_mpp, {"type", "rpc_establish_mpp_max"}),                                                          \
         F(type_active_threads_of_establish_mpp, {"type", "rpc_establish_mpp"}),                                                           \
         F(type_max_threads_of_dispatch_mpp, {"type", "rpc_dispatch_mpp_max"}),                                                            \

diff --git a/dbms/src/Common/UnaryCallback.h b/dbms/src/Common/UnaryCallback.h
@@ -0,0 +1,25 @@
+// Copyright 2022 PingCAP, Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#pragma once
+
+namespace DB
+{
+template <typename T>
+struct UnaryCallback
+{
+    virtual void execute(T & val) = 0;
+    virtual ~UnaryCallback() = default;
+};
+} // namespace DB
diff --git a/dbms/src/Debug/dbgFuncCoprocessor.cpp b/dbms/src/Debug/dbgFuncCoprocessor.cpp
@@ -389,7 +389,8 @@ BlockInputStreamPtr executeQuery(Context & context, RegionID region_id, const DA
                     root_tm,
                     context.getTMTContext().getKVCluster(),
                     context.getTMTContext().getMPPTaskManager(),
-                    context.getSettingsRef().enable_local_tunnel),
+                    context.getSettingsRef().enable_local_tunnel,
+                    context.getSettingsRef().enable_async_grpc_client),
                 tipb_exchange_receiver.encoded_task_meta_size(),
                 10,
                 nullptr);

diff --git a/dbms/src/Flash/Coprocessor/DAGContext.cpp b/dbms/src/Flash/Coprocessor/DAGContext.cpp
@@ -195,7 +195,8 @@ void DAGContext::initExchangeReceiverIfMPP(Context & context, size_t max_streams
                         getMPPTaskMeta(),
                         context.getTMTContext().getKVCluster(),
                         context.getTMTContext().getMPPTaskManager(),
-                        context.getSettingsRef().enable_local_tunnel),
+                        context.getSettingsRef().enable_local_tunnel,
+                        context.getSettingsRef().enable_async_grpc_client),
                     executor.exchange_receiver().encoded_task_meta_size(),
                     max_streams,
                     log);