fix a bug of client;

add performance test code.

examples/client/main.cpp

@@ -254,7 +254,61 @@ void test_sync_client() {
 	test_get_person();
 }
 
+void test_performance1() {
+	rpc_client client("127.0.0.1", 9000);
+	bool r = client.connect();
+	if (!r) {
+		std::cout << "connect timeout" << std::endl;
+		return;
+	}
+
+	person p{ 1, "tom", 20 };
+
+	for (size_t i = 0; i < 100000000; i++) {
+		auto future = client.async_call("get_name", p);
+		auto status = future.wait_for(2s);
+		if (status == std::future_status::deferred) {
+			std::cout << "deferred\n";
+		}
+		else if (status == std::future_status::timeout) {
+			std::cout << "timeout\n";
+		}
+		else if (status == std::future_status::ready) {
+		}
+	}
+
+	std::cout << "finish\n";
+
+	std::string str;
+	std::cin >> str;
+}
+
+void test_performance2() {
+	rpc_client client("127.0.0.1", 9000);
+	bool r = client.connect();
+	if (!r) {
+		std::cout << "connect timeout" << std::endl;
+		return;
+	}
+
+	person p{ 1, "tom", 20 };
+
+	try {
+		for (size_t i = 0; i < 100000000; i++) {
+			client.call<std::string>("get_name", p);
+		}
+		std::cout << "finish\n";
+	}
+	catch (const std::exception& ex) {
+		std::cout << ex.what() << '\n';
+	}
+
+	std::string str;
+	std::cin >> str;
+}
+
 int main() {
+	test_performance1();
 	test_sync_client();
 	test_async_client();
 	//test_upload();

examples/server/basic_server.vcxproj

@@ -47,7 +47,7 @@
   <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Release|x64'" Label="Configuration">
     <ConfigurationType>Application</ConfigurationType>
     <UseDebugLibraries>false</UseDebugLibraries>
-    <PlatformToolset>v141</PlatformToolset>
+    <PlatformToolset>v142</PlatformToolset>
     <WholeProgramOptimization>true</WholeProgramOptimization>
     <CharacterSet>MultiByte</CharacterSet>
   </PropertyGroup>
@@ -136,6 +136,9 @@
   <ItemGroup>
     <ClCompile Include="main.cpp" />
   </ItemGroup>
+  <ItemGroup>
+    <ClInclude Include="qps.h" />
+  </ItemGroup>
   <Import Project="$(VCTargetsPath)\Microsoft.Cpp.targets" />
   <ImportGroup Label="ExtensionTargets">
   </ImportGroup>

examples/server/basic_server.vcxproj.filters

@@ -19,4 +19,9 @@
       <Filter>源文件</Filter>
     </ClCompile>
   </ItemGroup>
+  <ItemGroup>
+    <ClInclude Include="qps.h">
+      <Filter>头文件</Filter>
+    </ClInclude>
+  </ItemGroup>
 </Project>

examples/server/main.cpp

@@ -2,6 +2,9 @@
 using namespace rest_rpc;
 using namespace rpc_service;
 #include <fstream>
+
+#include "qps.h"
+
 struct dummy{
 	int add(connection* conn, int a, int b) { return a + b; }
 };
@@ -57,6 +60,13 @@ std::string download(connection* conn, const std::string& filename) {
 	return content;
 }
 
+qps g_qps;
+
+std::string get_name(connection* conn, const person& p) {
+	g_qps.increase();
+	return p.name;
+}
+
 int main() {
 	rpc_server server(9000, 4);
 
@@ -68,6 +78,7 @@ int main() {
 	server.register_handler("get_person", get_person);
 	server.register_handler("upload", upload);
 	server.register_handler("download", download);
+	server.register_handler("get_name", get_name);
 
 	server.run();
 

examples/server/qps.h

@@ -0,0 +1,31 @@
+#pragma once
+#include <atomic>
+#include <thread>
+#include <chrono>
+
+class qps {
+public:
+	void increase() {
+		counter_.fetch_add(1, std::memory_order_release);
+	}
+
+	qps() {
+		thd_ = std::thread([this] {
+			while (!stop_) {
+				std::cout << "qps: " << counter_.load(std::memory_order_acquire) << '\n';
+				std::this_thread::sleep_for(std::chrono::seconds(1));
+				//counter_.store(0, std::memory_order_release);
+			}
+		});
+	}
+
+	~qps() {
+		stop_ = true;
+		thd_.join();
+	}
+
+private:
+	bool stop_ = false;
+	std::thread thd_;
+	std::atomic<uint32_t> counter_=0;
+};

include/router.h

@@ -175,7 +175,7 @@ class router : boost::noncopyable {
                                            std::placeholders::_5)};
   }
 
-  std::map<std::string,
+  std::unordered_map<std::string,
            std::function<void(connection*, const char*, size_t, std::string&, ExecMode& model)>>
       map_invokers_;
   std::function<void(const std::string&, std::string&&, connection*, bool)>

include/rpc_client.hpp

@@ -113,8 +113,9 @@ namespace rest_rpc {
 		template<typename T = void, typename... Args>
 		auto call(const std::string& rpc_name, Args&&... args) {
 			std::future<req_result> future = async_call(rpc_name, std::forward<Args>(args)...);
-			if (future.wait_for(std::chrono::seconds(wait_timeout_)) == std::future_status::timeout) {
-				throw std::out_of_range("timeout");
+			auto status = future.wait_for(std::chrono::seconds(1));
+			if (status == std::future_status::timeout || status == std::future_status::deferred) {
+				throw std::out_of_range("timeout or deferred");
 			}
 
 			if constexpr (std::is_void_v<T>) {
@@ -347,9 +348,12 @@ namespace rest_rpc {
 		}
 
 		std::future<req_result> get_future() {
-			auto p = std::promise<req_result>();
-			auto future = p.get_future();
-			future_map_.emplace(req_id_, std::move(p));
+			auto p = std::make_shared<std::promise<req_result>>();
+
+			std::future future = p->get_future();
+			strand_.post([this, p1 = std::move(p)] () mutable { 
+				future_map_.emplace(req_id_, std::move(p1)); 
+			});
 			return future;
 		}
 
@@ -365,8 +369,10 @@ namespace rest_rpc {
 			}
 			else {
 				auto& f = future_map_[req_id];
-				f.set_value(req_result{ data });
-				strand_.post([this, req_id]() { future_map_.erase(req_id); });
+				f->set_value(req_result{ data });
+				strand_.post([this, req_id]() { 
+					future_map_.erase(req_id); 
+				});
 			}
 		}
 
@@ -405,7 +411,7 @@ namespace rest_rpc {
 		std::function<void(boost::system::error_code)> err_cb_;
 
 		std::unordered_map<std::uint64_t, std::unique_ptr<call_t>> cb_map_;
-		std::unordered_map<std::uint64_t, std::promise<req_result>> future_map_;
+		std::unordered_map<std::uint64_t, std::shared_ptr<std::promise<req_result>>> future_map_;
 
 		char head_[HEAD_LEN] = {};
 		std::vector<char> body_;